package com.huitian.net.impl


import java.io.{BufferedReader, IOException, InputStreamReader}
import java.text.SimpleDateFormat
import java.util.Calendar

import ch.hsr.geohash.GeoHash
import com.alibaba.fastjson.{JSON, JSONObject}
import com.huitian.net.{MysqlConfiguration, SparkConfiguration}
import com.huitian.net.core.Adapter
import com.huitian.net.model.{TargetWeek, TargetWeekResult}
import com.huitian.net.utils.MysqlUtils
import org.apache.commons.logging.Log
import org.apache.log4j.Logger
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.sql.{DataFrame, Dataset, SaveMode, SparkSession}

import scala.collection.mutable.ArrayBuffer
import scala.util.control.Breaks

object EnAdapterImpl extends Adapter {
  override def get_dws_saleinvoice_in_salor_nc_m_hive(spark: SparkSession, day: String, lastday: String, numPartition: Int, log: Logger): Unit = {
    val dwd_m_sql =
      s"""
         |with temp1 as(
         |select
         |tjsyb,
         |sjbm,
         |fq,
         |pscode,
         |psnname,
         |SUM(nnum) AS current_num,
         |SUM(nmny) AS current_nmny,
         |sum(norigtaxnetprice) as norigtaxnetprice,
         |sum(norigmny) as norigmny,
         |sum(ntax) as ntax,
         |sum(norigtaxmny) as norigtaxmny,
         |sum(ntaxmny_price) as ntaxmny_price,
         |sum(ntaxmny) as ntaxmny,
         |sum(price_no_tax_currency) as price_no_tax_currency,
         |sum(ntotalincomenum) as ntotalincomenum,
         |sum(ntotalincomemny) as ntotalincomemny,
         |sum(ntotalpaymny) as ntotalpaymny,
         |sum(ntotalreceiv) as ntotalreceiv,
         |substring(month,0,4) as year,
         |month as month,
         |month as pt_month
         |from dwd.dwd_saleinvoice_nc
         |where pt_date>='$lastday'
         |and pt_date<='$day'
         |group by tjsyb,sjbm,fq,pscode,psnname,month
         |),temp2 as(
         |select
         |a.tjsyb as tjsyb,
         |a.sjbm as sjbm,
         |a.fq as fq,
         |a.pscode as pscode,
         |a.psnname as psnname,
         |a.current_num as current_num,
         |a.current_nmny as current_nmny,
         |b.current_num AS prev_year_num,
         |b.current_nmny AS prev_year_nmny,
         |c.current_num AS prev_month_num,
         |c.current_nmny AS prev_month_nmny,
         |a.norigtaxnetprice as norigtaxnetprice,
         |a.norigmny as norigmny,
         |a.ntax as ntax,
         |a.norigtaxmny as norigtaxmny,
         |a.ntaxmny_price as ntaxmny_price,
         |a.ntaxmny as ntaxmny,
         |a.price_no_tax_currency as price_no_tax_currency,
         |a.ntotalincomenum as ntotalincomenum,
         |a.ntotalincomemny as ntotalincomemny,
         |a.ntotalpaymny as ntotalpaymny,
         |a.ntotalreceiv as ntotalreceiv,
         |substring(a.month,0,4) as year,
         |a.month as month
         |from temp1 a left join temp1 b
         |on a.tjsyb=b.tjsyb
         |and a.sjbm=b.sjbm
         |and a.fq=b.fq
         |and a.pscode=b.pscode
         |and a.psnname=b.psnname
         |and b.month=CONCAT(CAST(SUBSTRING(a.month, 1, 4) AS INT) - 1, SUBSTRING(a.month, 5, 2))
         |left join temp1 c
         |on a.tjsyb=c.tjsyb
         |and a.sjbm=c.sjbm
         |and a.fq=c.fq
         |and a.pscode=c.pscode
         |and a.psnname=c.psnname
         |and c.month=date_format(add_months(from_unixtime(unix_timestamp(concat(a.month, '01'), 'yyyyMMdd')),-1),'yyyyMM')
         |)
         |insert overwrite table dws.dws_saleinvoice_in_salor_nc_m PARTITION (pt_month)
         |select
         |tjsyb,
         |sjbm,
         |fq,
         |pscode,
         |psnname,
         |current_num,
         |current_nmny,
         |prev_year_num,
         |prev_month_num,
         |prev_year_nmny,
         |prev_month_nmny,
         |current_num - prev_year_num as num_year_growth, --主数量同比增长额
         |current_num - prev_month_num AS num_month_growth, --主数量环比增长额
         |CASE WHEN prev_year_num = 0 OR prev_year_num IS NULL THEN NULL ELSE ROUND((current_num - prev_year_num) / prev_year_num * 100, 2)
         | END AS num_year_growth_rate, --主数量同比增长率
         |CASE WHEN prev_month_num = 0 OR prev_month_num IS NULL THEN NULL ELSE ROUND((current_num - prev_month_num) / prev_month_num * 100, 2)
         |END AS num_month_growth_rate, --主数量环比增长率
         |current_nmny - prev_year_nmny AS nmny_year_growth, --发货单本币未税金额同比增长额
         |current_nmny - prev_month_nmny AS nmny_month_growth, --发货单本币未税金额环比增长额
         |CASE WHEN prev_year_nmny = 0 OR prev_year_nmny IS NULL THEN NULL ELSE ROUND((current_nmny - prev_year_nmny) / prev_year_nmny * 100, 2)
         |END AS nmny_year_growth_rate, --发货单本币未税金额同比增长率
         |CASE WHEN prev_month_nmny = 0 OR prev_month_nmny IS NULL THEN NULL ELSE ROUND((current_nmny - prev_month_nmny) / prev_month_nmny * 100, 2)
         |END AS nmny_month_growth_rate, --发货单本币未税金额环比增长率
         |norigtaxnetprice,
         |norigmny,
         |ntax,
         |norigtaxmny,
         |ntaxmny_price,
         |ntaxmny,
         |price_no_tax_currency,
         |ntotalincomenum,
         |ntotalincomemny,
         |ntotalpaymny,
         |ntotalreceiv,
         |year,
         |CASE
         |-- 当提取的月份是 01 - 03 时，判定为一季度
         |WHEN substring(month, 5, 2) IN ('01', '02', '03') THEN '1'
         |-- 当提取的月份是 04 - 06 时，判定为二季度
         |WHEN substring(month, 5, 2) IN ('04', '05', '06') THEN '2'
         |-- 当提取的月份是 07 - 09 时，判定为三季度
         |WHEN substring(month, 5, 2) IN ('07', '08', '09') THEN '3'
         |-- 当提取的月份是 10 - 12 时，判定为四季度
         |WHEN substring(month, 5, 2) IN ('10', '11', '12') THEN '4'
         |-- 若月份不在上述范围内，返回 NULL
         |ELSE NULL
         |END AS quarter,
         |month as month,
         |CAST(SUBSTRING(month, 5, 2) AS INT) AS month2,
         |month as pt_month
         |from temp2
  """.stripMargin

    log.info("数据插入到 dws_saleinvoice_in_salor_nc_m 表")
    log.info("sql===="+dwd_m_sql)
    spark.sql(dwd_m_sql)



  }

  override def get_dws_saleinvoice_in_salor_nc_d_hive(spark: SparkSession, day: String, lastday: String, numPartition: Int, log: Logger): Unit = {
    val text =
      s"""
         |insert overwrite table dws.dws_saleinvoice_in_salor_nc_d PARTITION (pt_date)
         |select
         |tjsyb,
         |sjbm,
         |fq,
         |pscode,
         |psnname,
         |sum(nnum) as nnum,
         |sum(norigtaxnetprice) as norigtaxnetprice,
         |sum(norigmny) as norigmny,
         |sum(ntax) as ntax,
         |sum(norigtaxmny) as norigtaxmny,
         |sum(ntaxmny_price) as ntaxmny_price,
         |sum(ntaxmny) as ntaxmny,
         |sum(price_no_tax_currency) as price_no_tax_currency,
         |sum(nmny) as nmny,
         |sum(ntotalincomenum) as ntotalincomenum,
         |sum(ntotalincomemny) as ntotalincomemny,
         |sum(ntotalpaymny) as ntotalpaymny,
         |sum(ntotalreceiv) as ntotalreceiv,
         |substring(day,0,4) as year,
         |substring(day,0,6) as month,
         |CAST(SUBSTRING(day, 5, 2) AS INT) AS month2,
         |day as day,
         |CASE
         |-- 当提取的月份是 01 - 03 时，判定为一季度
         |WHEN substring(day, 5, 2) IN ('01', '02', '03') THEN '1'
         |-- 当提取的月份是 04 - 06 时，判定为二季度
         |WHEN substring(day, 5, 2) IN ('04', '05', '06') THEN '2'
         |-- 当提取的月份是 07 - 09 时，判定为三季度
         |WHEN substring(day, 5, 2) IN ('07', '08', '09') THEN '3'
         |-- 当提取的月份是 10 - 12 时，判定为四季度
         |WHEN substring(day, 5, 2) IN ('10', '11', '12') THEN '4'
         |-- 若月份不在上述范围内，返回 NULL
         |ELSE NULL
         |END AS quarter,
         |day as pt_date
         |from dwd.dwd_saleinvoice_nc
         |where pt_date>='$lastday'
         |and pt_date<='$day'
         |group by tjsyb,sjbm,fq,pscode,psnname,day

  """.stripMargin

    spark.sql(text)
    log.info("数据插入到 dws_saleinvoice_in_salor_nc_d 表")

  }

  override def get_dwd_saleinvoice_nc_hive(spark: SparkSession, current_year: String, last_year: String, numPartition: Int, log: Logger): Unit = {
    val text =
      s"""
         |with temp0 as(
         |--获取开票单子表数据
         |select
         |csaleinvoiceid,
         |cfirstid,
         |dbilldate,
         |vsrccode,
         |norigtaxmny,
         |csrcbid,
         |csaleorgid,
         |cordercustid,
         |cdeptid,
         |cemployeeid,
         |csendstordocid,
         |cmaterialvid,
         |csaleinvoicebid,
         |nnum,   --主数量
         |norigtaxnetprice, --主含税净价
         |norigmny, --无税金额
         |ntax,    --税额
         |blargessflag,
         |ntaxmny,
         |nmny,
         |ntotalincomenum,
         |ntotalincomemny,
         |ntotalpaymny,
         |ndiscountrate,
         |nqtprice,
         |norigdiscount,
         |nqttaxnetprice,
         |nitemdiscountrate,
         |nqttaxprice,
         |ntaxrate,
         |concat(substring(dbilldate,0,4),substring(dbilldate,6,2),substring(dbilldate,9,2)) as day,
         |dr,
         |row_number() over(partition by csaleinvoicebid  order by dbilldate desc) as rn
         |from ods.ods_so_saleinvoice_b_nc
         |where dr=0
         |AND ((pt_year<='${current_year}' AND pt_year>= '${last_year}'))
         |and concat(substring(dbilldate,0,4))<= '${current_year}'
         |and concat(substring(dbilldate,0,4))>= '${last_year}'
         |),temp01 as(
         |--子表数据去重
         |select
         |csaleinvoiceid,
         |cfirstid,
         |dbilldate,
         |vsrccode,
         |norigtaxmny,
         |csrcbid,
         |csaleorgid,
         |cordercustid,
         |cdeptid,
         |cemployeeid,
         |csendstordocid,
         |cmaterialvid,
         |csaleinvoicebid,
         |nnum,
         |norigtaxnetprice,
         |norigmny,
         |ntax,
         |blargessflag,
         |ntaxmny,
         |nmny,
         |ntotalincomenum,
         |ntotalincomemny,
         |ntotalpaymny,
         |ndiscountrate,
         |nqtprice,
         |norigdiscount,
         |nqttaxnetprice,
         |nitemdiscountrate,
         |nqttaxprice,
         |ntaxrate,
         |day,
         |dr
         |from temp0
         |where rn=1
         |),temp1 as(
         |--解决荣盛和越友 的汇率问题
         |select
         |temp01.csaleinvoiceid as csaleinvoiceid,
         |temp01.cfirstid as cfirstid,
         |temp01.dbilldate as dbilldate,
         |temp01.vsrccode as vsrccode,
         |temp01.norigtaxmny as norigtaxmny,
         |temp01.csrcbid as csrcbid,
         |temp01.csaleorgid as csaleorgid,
         |temp01.cordercustid as cordercustid,
         |temp01.cdeptid as cdeptid,
         |temp01.cemployeeid as cemployeeid,
         |temp01.csendstordocid as csendstordocid,
         |temp01.cmaterialvid as cmaterialvid,
         |temp01.csaleinvoicebid as csaleinvoicebid,
         |temp01.nnum as nnum,
         |temp01.norigtaxnetprice as norigtaxnetprice,
         |temp01.norigmny as norigmny,
         |temp01.ntax as ntax,
         |temp01.blargessflag as blargessflag,
         |temp01.ntaxmny as ntaxmny,
         |(case when j.name like '%荣盛新材%' or j.name like '%越友有限%' then temp01.nmny/3500 else temp01.nmny end) as nmny,
         |temp01.ntotalincomenum as ntotalincomenum,
         |temp01.ntotalincomemny as ntotalincomemny,
         |temp01.ntotalpaymny as ntotalpaymny,
         |temp01.day as day,
         |temp01.dr as dr
         |from temp01
         |LEFT JOIN dim.dim_org_salesorg_nc j ON temp01.csaleorgid = j.pk_salesorg
         |),temp2 as (
         |--获取开票单表头数据
         |select
         |csaleinvoiceid,
         |cinvoicecustid,
         |vbillcode,
         |dbilldate,
         |pk_org,
         |vdef4,
         |concat(substring(dbilldate,0,4),substring(dbilldate,6,2),substring(dbilldate,9,2)) as day,
         |row_number() over(partition by csaleinvoiceid ,dbilldate order by dbilldate desc) as rn,
         |dr
         |from ods.ods_so_saleinvoice_nc
         |where dr=0
         |AND ((pt_year <= '${current_year}' AND pt_year >= '${last_year}'))
         |and concat(substring(dbilldate,0,4))<= '${current_year}'
         |and concat(substring(dbilldate,0,4))>= '${last_year}'
         |),temp21 as(
         |--开票单表头数据去重
         |select
         |csaleinvoiceid,
         |cinvoicecustid,
         |vbillcode,
         |dbilldate,
         |pk_org,
         |vdef4,
         |day,
         |dr
         |from temp2
         |where rn=1
         |),temp3 as (
         |--获取维度表数据：注意这一块维度可能存在问题，一个员工多个部门的问题
         |select  a.pk_org,a.pk_customer,a.pk_custsaleclass, b.name fq,c.name sjbm ,d.name tjsyb
         |,e.code pscode,e.name psnname,f.name qdlx,g.pk_salesorg,a.pk_financeorg
         |from dim.dim_bd_custsale_nc a
         |left join dim.dim_bd_custsaleclass_nc b on a.pk_custsaleclass=b.pk_custsaleclass
         |left join dim.dim_bd_custsaleclass_nc c on b.pk_father =c.pk_custsaleclass
         |left join dim.dim_bd_custsaleclass_nc d on substr(b.code,1,2)=d.code
         |left join dim.dim_bd_psndoc_nc e on a.respperson =e.pk_psndoc
         |left join dim.dim_bd_channeltype_nc f on a.channel=f.pk_type
         |left join dim.dim_org_salesorg_nc g on d.name=g.name
         |where a.pk_custsaleclass<>'~'
         |), tem_order1 as(
         |--获取销量订单表数据，过滤vdef4相关
         |select
         |vdef4,
         |dbilldate,
         |dr,
         |csaleorderid
         |from(
         |select
         |csaleorderid,
         |vdef4,
         |dbilldate,
         |dr,
         |row_number() over(partition by csaleorderid  order by dbilldate desc) as rn
         |from ods.ods_so_saleorder_nc
         |where dr=0
         |AND ((pt_year<='${current_year}' AND pt_year>= '${last_year}') or pt_year='his')
         |and concat(substring(dbilldate,0,4))<= '${current_year}'
         |and concat(substring(dbilldate,0,4))>= '${last_year}'
         |) t
         |where t.rn=1
         |),view_fpfl as (
         |select csaleinvoicebid
         |,round((1-a.ndiscountrate/100)*a.nnum*a.nqtprice,2) yje
         |,round((norigdiscount-(1-a.nitemdiscountrate /100)*a.nnum*a.nqttaxprice)/(1+ntaxrate/100),2) jee
         |,round((a.nnum*a.nqttaxprice-a.nnum*a.nqttaxnetprice-(1-a.nitemdiscountrate/100)*a.nnum*a.nqttaxprice) /(1+ntaxrate/100),2) je
         |from temp01 a
         |left join tem_order1 b
         |on a.cfirstid =b.csaleorderid
         |where b.vdef4='1001A11000000000058S'
         |and a.dr=0
         |),temp_r as (
         |select
         |a.csaleinvoiceid as csaleinvoiceid,
         |a.csaleorgid  as csaleorgid,
         |a.csaleinvoicebid as csaleinvoicebid,
         |b.vbillcode as vbillcode, --发票号
         |a.vsrccode as vsrccode, --来源单据号
         |b.dbilldate as dbilldate, --发票日期
         |h.code as customer_code, --客户编码
         |h.name as customer_name, --客户名称
         |ii.code invoice_code,--发票客户编码
         |ii.name invoice_name,--发票客户名称
         |g.qdlx customer_type,--客户类型
         |j.name as org_salesorg_name,--销售组织
         |k.name as org_financeorg_name,--财务组织
         |g.tjsyb as tjsyb,--事业部
         |g.sjbm as sjbm, --销售部门
         |n.name as storehouse,--仓库
         |g.fq as fq,--片区
         |g.pscode as pscode,--业务员工号
         |g.psnname as psnname,--业务员
         |t.name as research_team ,--课题组
         |uu.name as material_industry ,--行业
         |nvl(v.name,t.name) as research_institute ,--研究院
         |o.code as material_code, --物料编码
         |o.name as material_name,--物料名称
         |o.materialspec as material_spec, --物料规格
         |p.def1 as material_df1,--物料大类
         |nvl(a.nnum,0) as nnum, --主数量
         |nvl(a.norigtaxnetprice,0) as norigtaxnetprice, --主含税净价
         |nvl(a.norigmny,0) as norigmny, --无税金额
         |nvl(a.ntax,0) as ntax, --税额
         |(case when a.blargessflag = 'n' then nvl(a.norigtaxmny, 0) else 0 end) as norigtaxmny, --价税合计
         |(case when nvl(a.nnum,0)=0 then 0 else nvl(a.ntaxmny,0)/nvl(a.nnum,0) end) as ntaxmny_price,--本币含税单价
         |nvl(a.ntaxmny,0) as ntaxmny, --本币价税合计
         |(case when nvl(a.nnum,0)=0 then 0 else nvl(a.nmny,0)/nvl(a.nnum,0) end)  as price_no_tax_currency, --本币无税单价
         |nvl(a.nmny,0)+nvl(q.je,0) as nmny ,--本币无税金额
         |nvl(a.ntotalincomenum,0) as ntotalincomenum , --应收数量
         |nvl(a.ntotalincomemny,0) as ntotalincomemny , --应收金额
         |nvl(a.ntotalpaymny,0) as ntotalpaymny ,--收款金额
         |nvl(a.ntotalincomemny,0)-nvl(a.ntotalpaymny,0) as ntotalreceiv,   --应收余额
         |substring(a.day,0,4) as year,
         |CASE
         |-- 当提取的月份是 01 - 03 时，判定为一季度
         |WHEN substring(a.day, 5, 2) IN ('01', '02', '03') THEN '1'
         |-- 当提取的月份是 04 - 06 时，判定为二季度
         |WHEN substring(a.day, 5, 2) IN ('04', '05', '06') THEN '2'
         |-- 当提取的月份是 07 - 09 时，判定为三季度
         |WHEN substring(a.day, 5, 2) IN ('07', '08', '09') THEN '3'
         |-- 当提取的月份是 10 - 12 时，判定为四季度
         |WHEN substring(a.day, 5, 2) IN ('10', '11', '12') THEN '4'
         |-- 若月份不在上述范围内，返回 NULL
         |ELSE NULL
         |END AS quarter,
         |substring(a.day,0,6) as month,
         |cast(substring(a.day, 5, 2) as int) as month2,
         |(floor((dayofmonth(from_unixtime(unix_timestamp(a.day, 'yyyyMMdd'), 'yyyy-MM-dd')) - 1) / 7) + 1) AS week,
         |a.day as day,
         |a.day as pt_date,
         |row_number() over(partition by a.csaleinvoicebid ,a.dbilldate order by a.dbilldate desc) as rn
         |FROM temp1 a --开票体
         |LEFT JOIN temp21 b ON a.csaleinvoiceid = b.csaleinvoiceid and a.day=b.day --开票头
         |LEFT JOIN temp3 g ON a.csaleorgid = g.pk_org
         |AND g.pk_customer = a.cordercustid --客户类型
         |LEFT JOIN dim.dim_bd_customer_nc h ON a.cordercustid = h.pk_customer --订单客户表
         |LEFT JOIN dim.dim_bd_customer_nc ii ON b.cinvoicecustid = ii.pk_customer --发票客户表
         |LEFT JOIN dim.dim_org_salesorg_nc j ON a.csaleorgid = j.pk_salesorg --销售组织
         |LEFT JOIN dim.dim_org_financeorg_nc k ON b.pk_org = k.pk_financeorg --财务组织
         |LEFT JOIN dim.dim_org_dept_nc l ON a.cdeptid = l.pk_dept --部门
         |LEFT JOIN dim.dim_bd_psndoc_nc m ON a.cemployeeid = m.pk_psndoc --业务员
         |LEFT JOIN dim.dim_bd_stordoc_nc n ON a.csendstordocid = n.pk_stordoc --仓库
         |LEFT JOIN dim.dim_bd_material_v_nc o ON a.cmaterialvid = o.pk_source --物料表
         |LEFT JOIN dim.dim_bd_marbasclass_nc p ON substr (o.code, 1, 3) = p.code --物料大类
         |LEFT JOIN view_fpfl q ON a.csaleinvoicebid = q.csaleinvoicebid --返利金额
         |left join dim.dim_bd_defdoc_nc t on o.def5 = t.pk_defdoc --课题组
         |left join dim.dim_bd_defdoc_nc uu on o.def12 = uu.pk_defdoc --行业
         |left join dim.dim_bd_defdoc_nc v on uu.pid=v.pk_defdoc  --研究院
         |),temp4 as(
         |select
         |csaleinvoiceid,
         |csaleorgid,
         |csaleinvoicebid,
         |vbillcode,
         |vsrccode,
         |dbilldate,
         |customer_code,
         |customer_name,
         |invoice_code,
         |invoice_name,
         |customer_type,
         |org_salesorg_name,
         |org_financeorg_name,
         |tjsyb,
         |sjbm,
         |storehouse,
         |fq,
         |pscode,
         |psnname,
         |research_team ,
         |material_industry ,
         |research_institute ,
         |material_code,
         |material_name,
         |material_spec,
         |material_df1,
         |nnum,
         |norigtaxnetprice,
         |norigmny,
         |ntax,
         |norigtaxmny,
         |ntaxmny_price,
         |ntaxmny,
         |price_no_tax_currency,
         |nmny ,
         |ntotalincomenum ,
         |ntotalincomemny ,
         |ntotalpaymny ,
         |ntotalreceiv,
         |year,
         |quarter,
         |month,
         |month2,
         |week,
         |day,
         |pt_date
         |FROM temp_r
         |where rn=1
         |),temp5 as(
         |select
         |a.code as material_code,
         |a.name as material_name,
         |a.creationtime as creationtime,
         |CASE WHEN a.enablestate = 1 THEN '未启用' WHEN a.enablestate = 2 THEN '已启用' WHEN a.enablestate = 3 THEN '已停用' ELSE NULL END AS status,
         |a.materialshortname as materialshortname,
         |a.materialspec as materialspec,
         |b.def1 as material_def1,
         |c.name as material_series,
         |e1.name as material_class1,
         |e2.name as material_class2,
         |e3.name as material_industry1,
         |e4.name as material_industry2,
         |e5.name as research_team,
         |g.name as group_name,
         |h.name as place_of_origin
         |from dim.dim_bd_material_v_nc a
         |left join dim.dim_bd_marbasclass_nc b on substr(a.code,1,3)=b.code
         |left join dim.dim_bd_marbasclass_nc c on a.pk_marbasclass=c.pk_marbasclass
         |left join dim.dim_bd_defdoc_nc e1 on  a.def13 = e1.pk_defdoc
         |left join dim.dim_bd_defdoc_nc e2 on  a.def11 = e2.pk_defdoc
         |left join dim.dim_bd_defdoc_nc e3 on  a.def12 = e3.pk_defdoc
         |left join dim.dim_bd_defdoc_nc e4 on  a.def1 = e4.pk_defdoc
         |left join dim.dim_bd_defdoc_nc e5 on  a.def5 = e5.pk_defdoc
         |left join dim.dim_org_group_nc g on  a.pk_group=g.pk_group
         |left join dim.dim_bd_defdoc_nc h on  a.prodarea =h.pk_defdoc
         |),temp6 as(
         |select
         |temp4.csaleinvoiceid as csaleinvoiceid,
         |temp4.csaleorgid as csaleorgid,
         |temp4.csaleinvoicebid as csaleinvoicebid,
         |temp4.vbillcode as vbillcode,
         |temp4.vsrccode as vsrccode,
         |temp4.dbilldate as dbilldate,
         |temp4.customer_code as customer_code,
         |temp4.customer_name as customer_name,
         |temp4.invoice_code as invoice_code,
         |temp4.invoice_name as invoice_name,
         |temp4.customer_type as customer_type,
         |temp4.org_salesorg_name as org_salesorg_name,
         |temp4.org_financeorg_name as org_financeorg_name,
         |temp4.tjsyb as tjsyb,
         |temp4.sjbm as sjbm,
         |temp4.storehouse as storehouse,
         |temp4.fq as fq,
         |temp4.pscode as pscode,
         |temp4.psnname as psnname,
         |temp4.research_team as research_team,
         |temp4.material_industry as material_industry,
         |temp4.research_institute as research_institute,
         |temp4.material_code as material_code,
         |temp4.material_name as material_name,
         |temp4.material_spec as material_spec,
         |temp4.material_df1 as material_df1,
         |temp5.material_series as material_series,
         |temp5.material_class1 as material_class1,
         |temp5.material_class2 as material_class2,
         |temp5.material_industry1 as material_industry1,
         |temp5.material_industry2 as material_industry2,
         |temp4.nnum as nnum,
         |temp4.norigtaxnetprice as norigtaxnetprice,
         |temp4.norigmny as norigmny,
         |temp4.ntax as ntax,
         |temp4.norigtaxmny as norigtaxmny,
         |temp4.ntaxmny_price as ntaxmny_price,
         |temp4.ntaxmny as ntaxmny,
         |temp4.price_no_tax_currency as price_no_tax_currency,
         |temp4.nmny as nmny,
         |temp4.ntotalincomenum as ntotalincomenum,
         |temp4.ntotalincomemny as ntotalincomemny,
         |temp4.ntotalpaymny as ntotalpaymny,
         |temp4.ntotalreceiv as ntotalreceiv,
         |temp4.year as year,
         |temp4.quarter as quarter,
         |temp4.month as month,
         |temp4.month2 as month2,
         |temp4.week as week,
         |temp4.day as day,
         |temp4.pt_date as pt_date
         |from temp4 left join temp5
         |on temp4.material_code=temp5.material_code
         |),temp7 as(
         |select
         |day,
         |week
         |from dim.dim_time_in_week
         |)
         |INSERT OVERWRITE dwd.dwd_saleinvoice_nc PARTITION (pt_date)
         |select
         |csaleinvoiceid,
         |csaleorgid,
         |csaleinvoicebid,
         |vbillcode,
         |vsrccode,
         |dbilldate,
         |customer_code,
         |customer_name,
         |invoice_code,
         |invoice_name,
         |customer_type,
         |org_salesorg_name,
         |org_financeorg_name,
         |tjsyb,
         |sjbm,
         |storehouse,
         |fq,
         |pscode,
         |psnname,
         |research_team,
         |material_industry,
         |research_institute,
         |material_code,
         |material_name,
         |material_spec,
         |material_df1,
         |material_series,
         |material_class1,
         |material_class2,
         |material_industry1,
         |material_industry2,
         |nnum,
         |norigtaxnetprice,
         |norigmny,
         |ntax,
         |norigtaxmny,
         |ntaxmny_price,
         |ntaxmny,
         |price_no_tax_currency,
         |(case when org_financeorg_name like '%荣盛新材%' or org_financeorg_name like '%越友有限%' then nmny/3500 else nmny end) as nmny,
         |ntotalincomenum,
         |ntotalincomemny,
         |ntotalpaymny,
         |ntotalreceiv,
         |year,
         |quarter,
         |month,
         |month2,
         |nvl(temp7.week,temp6.week) as week,
         |temp6.day as day,
         |pt_date
         |from temp6 left join temp7
         |on temp6.day=temp7.day
  """.stripMargin

    spark.sql(text)
    log.info("数据插入到 dwd_saleinvoice_nc 表")


  }

  override def dws_delivery_in_salor_nc_m_hive(spark: SparkSession, day: String, lastday: String, numPartition: Int, log: Logger): Unit = {
    val text =
      s"""
         |with temp1 as(
         |SELECT
         |tjsyb,
         |sjbm,
         |fq,
         |pscode,
         |psnname,
         |month,
         |SUM(num) AS current_num,
         |SUM(nmny) AS current_nmny,
         | -- 前月数值（环比基准）
         |LAG(SUM(num), 1) OVER (PARTITION BY tjsyb, sjbm, fq, pscode, psnname ORDER BY month) AS prev_month_num,
         |LAG(SUM(nmny), 1) OVER (PARTITION BY tjsyb, sjbm, fq, pscode, psnname ORDER BY month) AS prev_month_nmny,
         |-- 去年同期数值（同比基准）
         |LAG(SUM(num), 12) OVER (PARTITION BY tjsyb, sjbm, fq, pscode, psnname ORDER BY month) AS prev_year_num,
         |LAG(SUM(nmny), 12) OVER (PARTITION BY tjsyb, sjbm, fq, pscode, psnname ORDER BY month) AS prev_year_nmny,
         |sum(norigdiscount) as norigdiscount
         |,sum(nqtorigtaxprice) as nqtorigtaxprice
         |,sum(nqtorigtaxnetprc) as nqtorigtaxnetprc
         |,sum(nqtorignetprice) as nqtorignetprice
         |,sum(norigmny) as norigmny
         |,sum(norigtaxmny) as norigtaxmny
         |,sum(nqttaxnetprice) as nqttaxnetprice
         |,sum(nqtnetprice) as nqtnetprice
         |,sum(ntaxmny) as ntaxmny
         |,sum(rebate_discount_amount) as rebate_discount_amount
         |,sum(local_currency_delivery) as local_currency_delivery
         |,count(vbillcode) as total_deliver_order
         |,COUNT(CASE WHEN vbillstatue = '自由' THEN vbillcode END) AS free_count
         |,COUNT(CASE WHEN vbillstatue = '审批通过' THEN vbillcode END) AS approved_count
         |,COUNT(CASE WHEN vbillstatue = '冻结' THEN vbillcode END) AS frozen_count
         |,COUNT(CASE WHEN vbillstatue = '关闭' THEN vbillcode END) AS closed_count
         |,COUNT(CASE WHEN vbillstatue = '审批中' THEN vbillcode END) AS approving_count
         |,COUNT(CASE WHEN vbillstatue = '审批不通过' THEN vbillcode END) AS rejected_count
         |,COUNT(CASE WHEN vbillstatue = '失效' THEN vbillcode END) AS invalid_count
         |FROM dwd.dwd_delivery_nc
         |where pt_date>='$lastday'
         |and pt_date<='$day'
         |GROUP BY tjsyb, sjbm, fq, pscode, psnname, month
         |),temp2 as(
         |select
         |a.tjsyb as tjsyb,
         |a.sjbm as sjbm,
         |a.fq as fq,
         |a.pscode as pscode,
         |a.psnname as psnname,
         |a.current_num,
         |a.current_nmny,
         |c.current_num AS prev_month_num
         |,c.current_nmny AS prev_month_nmny
         |,b.current_num AS prev_year_num
         |,b.current_nmny AS prev_year_nmny
         |,a.norigdiscount as norigdiscount
         |,a.nqtorigtaxprice as nqtorigtaxprice
         |,a.nqtorigtaxnetprc as nqtorigtaxnetprc
         |,a.nqtorignetprice as nqtorignetprice
         |,a.norigmny as norigmny
         |,a.norigtaxmny as norigtaxmny
         |,a.nqttaxnetprice as nqttaxnetprice
         |,a.nqtnetprice as nqtnetprice
         |,a.ntaxmny as ntaxmny
         |,a.rebate_discount_amount as rebate_discount_amount
         |,a.local_currency_delivery as local_currency_delivery
         |,a.total_deliver_order as total_deliver_order
         |,a.free_count as free_count
         |,a.approved_count as approved_count
         |,a.frozen_count as frozen_count
         |,a.closed_count as closed_count
         |,a.approving_count as approving_count
         |,a.rejected_count as rejected_count
         |,a.invalid_count as invalid_count
         |,a.month  as month
         |from temp1 a left join temp1 b
         |on a.tjsyb=b.tjsyb
         |and a.sjbm=b.sjbm
         |and a.fq=b.fq
         |and a.pscode=b.pscode
         |and a.psnname=b.psnname
         |and b.month=CONCAT(CAST(SUBSTRING(a.month, 1, 4) AS INT) - 1, SUBSTRING(a.month, 5, 2))
         |left join temp1 c
         |on a.tjsyb=c.tjsyb
         |and a.sjbm=c.sjbm
         |and a.fq=c.fq
         |and a.pscode=c.pscode
         |and a.psnname=c.psnname
         |and c.month=CONCAT(CAST(SUBSTRING(a.month, 1, 4) AS INT) - 1, SUBSTRING(a.month, 5, 2))
         |)
         |insert overwrite table dws.dws_delivery_in_salor_nc_m PARTITION (pt_month)
         |select
         |tjsyb,
         |sjbm,
         |fq,
         |pscode,
         |psnname,
         |current_num,
         |current_nmny,
         |prev_month_num
         |,prev_month_nmny
         |,prev_year_num
         |,prev_year_nmny
         |,current_num - prev_year_num as num_year_growth, --主数量同比增长额
         |current_num - prev_month_num AS num_month_growth, --主数量环比增长额
         |CASE WHEN prev_year_num = 0 OR prev_year_num IS NULL THEN NULL ELSE ROUND((current_num - prev_year_num) / prev_year_num * 100, 2)
         | END AS num_year_growth_rate, --主数量同比增长率
         |CASE WHEN prev_month_num = 0 OR prev_month_num IS NULL THEN NULL ELSE ROUND((current_num - prev_month_num) / prev_month_num * 100, 2)
         |END AS num_month_growth_rate, --主数量环比增长率
         |current_nmny - prev_year_nmny AS nmny_year_growth, --发货单本币未税金额同比增长额
         |current_nmny - prev_month_nmny AS nmny_month_growth, --发货单本币未税金额环比增长额
         |CASE WHEN prev_year_nmny = 0 OR prev_year_nmny IS NULL THEN NULL ELSE ROUND((current_nmny - prev_year_nmny) / prev_year_nmny * 100, 2)
         |END AS nmny_year_growth_rate, --发货单本币未税金额同比增长率
         |CASE WHEN prev_month_nmny = 0 OR prev_month_nmny IS NULL THEN NULL ELSE ROUND((current_nmny - prev_month_nmny) / prev_month_nmny * 100, 2)
         |END AS nmny_month_growth_rate --发货单本币未税金额环比增长率
         |,norigdiscount
         |,nqtorigtaxprice
         |,nqtorigtaxnetprc
         |,nqtorignetprice
         |,norigmny
         |,norigtaxmny
         |,nqttaxnetprice
         |,nqtnetprice
         |,ntaxmny
         |,rebate_discount_amount
         |,local_currency_delivery
         |,total_deliver_order
         |,free_count
         |,approved_count
         |,frozen_count
         |,closed_count
         |,approving_count
         |,rejected_count
         |,invalid_count
         |,substring(month,0,4) as year
         |,CASE
         |-- 当提取的月份是 01 - 03 时，判定为一季度
         |WHEN substring(month, 5, 2) IN ('01', '02', '03') THEN '1'
         |-- 当提取的月份是 04 - 06 时，判定为二季度
         |WHEN substring(month, 5, 2) IN ('04', '05', '06') THEN '2'
         |-- 当提取的月份是 07 - 09 时，判定为三季度
         |WHEN substring(month, 5, 2) IN ('07', '08', '09') THEN '3'
         |-- 当提取的月份是 10 - 12 时，判定为四季度
         |WHEN substring(month, 5, 2) IN ('10', '11', '12') THEN '4'
         |-- 若月份不在上述范围内，返回 NULL
         |ELSE NULL
         |END AS quarter
         |,month
         |,cast(substring(month, 5, 2) as int) as month2
         |,month as pt_month
         |from temp2
  """.stripMargin

    spark.sql(text)
    log.info("数据插入到 dws_delivery_in_salor_nc_m 表")

  }

  override def dws_delivery_in_salor_nc_d_hive(spark: SparkSession, day: String, lastday: String, numPartition: Int, log: Logger): Unit = {
    val text =
      s"""
         |with temp1 as(
         |select
         |tjsyb
         |,sjbm
         |,fq
         |,pscode
         |,psnname
         |,sum(num) as num
         |,sum(nmny) as nmny
         |,sum(norigdiscount) as norigdiscount
         |,sum(nqtorigtaxprice) as nqtorigtaxprice
         |,sum(nqtorigtaxnetprc) as nqtorigtaxnetprc
         |,sum(nqtorignetprice) as nqtorignetprice
         |,sum(norigmny) as norigmny
         |,sum(norigtaxmny) as norigtaxmny
         |,sum(nqttaxnetprice) as nqttaxnetprice
         |,sum(nqtnetprice) as nqtnetprice
         |,sum(ntaxmny) as ntaxmny
         |,sum(rebate_discount_amount) as rebate_discount_amount
         |,sum(local_currency_delivery) as local_currency_delivery
         |,count(vbillcode) as total_deliver_order
         |,COUNT(CASE WHEN vbillstatue = '自由' THEN vbillcode END) AS free_count
         |,COUNT(CASE WHEN vbillstatue = '审批通过' THEN vbillcode END) AS approved_count
         |,COUNT(CASE WHEN vbillstatue = '冻结' THEN vbillcode END) AS frozen_count
         |,COUNT(CASE WHEN vbillstatue = '关闭' THEN vbillcode END) AS closed_count
         |,COUNT(CASE WHEN vbillstatue = '审批中' THEN vbillcode END) AS approving_count
         |,COUNT(CASE WHEN vbillstatue = '审批不通过' THEN vbillcode END) AS rejected_count
         |,COUNT(CASE WHEN vbillstatue = '失效' THEN vbillcode END) AS invalid_count
         |,substring(day,0,4) as year
         |,CASE
         |-- 当提取的月份是 01 - 03 时，判定为一季度
         |WHEN substring(month, 5, 2) IN ('01', '02', '03') THEN '1'
         |-- 当提取的月份是 04 - 06 时，判定为二季度
         |WHEN substring(month, 5, 2) IN ('04', '05', '06') THEN '2'
         |-- 当提取的月份是 07 - 09 时，判定为三季度
         |WHEN substring(month, 5, 2) IN ('07', '08', '09') THEN '3'
         |-- 当提取的月份是 10 - 12 时，判定为四季度
         |WHEN substring(month, 5, 2) IN ('10', '11', '12') THEN '4'
         |-- 若月份不在上述范围内，返回 NULL
         |ELSE NULL
         |END AS quarter
         |,month
         |,cast(substring(month, 5, 2) as int) as month2
         |,(floor((dayofmonth(from_unixtime(unix_timestamp(day, 'yyyyMMdd'), 'yyyy-MM-dd')) - 1) / 7) + 1) AS week
         |,day
         |,day as pt_date
         |from dwd.dwd_delivery_nc
         |where pt_date>='$lastday'
         |and pt_date<='$day'
         |group by tjsyb,sjbm,fq,pscode,psnname,day,month,substring(day,0,4)
         |),temp2 as(
         |select
         |day,
         |week
         |from dim.dim_time_in_week
         |)
         |insert overwrite table dws.dws_delivery_in_salor_nc_d PARTITION (pt_date)
         |select
         |tjsyb
         |,sjbm
         |,fq
         |,pscode
         |,psnname
         |,num
         |,nmny
         |,norigdiscount
         |,nqtorigtaxprice
         |,nqtorigtaxnetprc
         |,nqtorignetprice
         |,norigmny
         |,norigtaxmny
         |,nqttaxnetprice
         |,nqtnetprice
         |,ntaxmny
         |,rebate_discount_amount
         |,local_currency_delivery
         |,total_deliver_order
         |,free_count
         |,approved_count
         |,frozen_count
         |,closed_count
         |,approving_count
         |,rejected_count
         |,invalid_count
         |,year
         |,quarter
         |,month
         |,month2
         |,nvl(temp1.week,temp2.week) as week
         |,temp1.day as day
         |,temp1.day as pt_date
         |from temp1 left join temp2
         |on temp1.day=temp2.day
  """.stripMargin

    spark.sql(text)
    log.info("数据插入到 dws_delivery_in_salor_nc_d 表")

  }
  //生成发货单dwd层数据  ,
  override def get_dwd_delivery_nc_hive(spark: SparkSession, current_year: String, last_year: String, numPartition: Int, log: Logger): Unit = {

    val text =
      s"""
         |with temp1 as(
         |--获取发货单主表数据
         |select
         |vbillcode,
         |fstatusflag,
         |vdef4,
         |cdeliveryid,
         |pk_group,
         |dr,
         |day
         |from(
         |select
         |vbillcode,
         |fstatusflag,
         |vdef4,
         |cdeliveryid,
         |pk_group,
         |dr,
         |row_number() over(partition by cdeliveryid   order by dbilldate desc) as rn ,
         |concat(substring(dbilldate,0,4),substring(dbilldate,6,2),substring(dbilldate,9,2)) as day
         |from ods.ods_so_delivery_nc
         |where dr=0
         |AND ((pt_year <= '${current_year}' AND pt_year >= '${last_year}'))
         |and concat(substring(dbilldate,0,4))<= '${current_year}'
         |and concat(substring(dbilldate,0,4))>= '${last_year}'
         |) t
         |where rn=1
         |),temp0 as(
         |--获取发货单子表数据
         |select
         |dbilldate,
         |cinvoicecustid,
         |cdeliveryid,
         |cmaterialvid,
         |cordercustid,
         |csaleorgid,
         |cdeptid,
         |cemployeeid,
         |csendstordocid,
         |cvendorid,
         |pk_group,
         |pk_org,
         |cdeliverybid,
         |csettleorgid,
         |crprofitcenterid,
         |dsenddate,
         |dr,
         |vsrccode,
         |nnum,
         |nmny,
         |norigdiscount,
         |nqtorigtaxprice,
         |nqtorigtaxnetprc,
         |nqtorignetprice,
         |norigmny,
         |norigtaxmny,
         |nqttaxnetprice,
         |nqtnetprice,
         |ntaxmny,
         |nitemdiscountrate,
         |nqttaxprice,
         |ntaxrate,
         |day
         |from(
         |select
         |dbilldate,
         |cinvoicecustid,
         |cdeliveryid,
         |cmaterialvid,
         |cordercustid,
         |csaleorgid,
         |cdeptid,
         |cemployeeid,
         |csendstordocid,
         |cvendorid,
         |pk_group,
         |pk_org,
         |cdeliverybid,
         |csettleorgid,
         |crprofitcenterid,
         |dsenddate,
         |dr,
         |vsrccode,
         |nnum,
         |nmny,
         |norigdiscount,
         |nqtorigtaxprice,
         |nqtorigtaxnetprc,
         |nqtorignetprice,
         |norigmny,
         |norigtaxmny,
         |nqttaxnetprice,
         |nqtnetprice,
         |ntaxmny,
         |nitemdiscountrate,
         |nqttaxprice,
         |ntaxrate,
         |row_number() over(partition by cdeliverybid   order by dsenddate desc) as rn,
         |concat(substring(dsenddate,0,4),substring(dsenddate,6,2),substring(dsenddate,9,2)) as day
         |from ods.ods_so_delivery_b_nc
         |where dr=0
         |AND ((pt_year <= '${current_year}' AND pt_year >= '${last_year}'))
         |and concat(substring(dsenddate,0,4))<= '${current_year}'
         |and concat(substring(dsenddate,0,4))>= '${last_year}'
         |)t
         |where rn=1
         |),temp2 as(
         |--解决：荣盛新材，越友有限 的货币汇率问题：发现不生效
         |select
         |temp0.dbilldate as dbilldate,
         |temp0.cinvoicecustid as cinvoicecustid,
         |temp0.cdeliveryid as cdeliveryid,
         |temp0.cmaterialvid as cmaterialvid,
         |temp0.cordercustid as cordercustid,
         |temp0.csaleorgid as csaleorgid,
         |temp0.cdeptid as cdeptid,
         |temp0.cemployeeid as cemployeeid,
         |temp0.csendstordocid as csendstordocid,
         |temp0.cvendorid as cvendorid,
         |temp0.pk_group as pk_group,
         |temp0.pk_org as pk_org,
         |temp0.cdeliverybid as cdeliverybid,
         |temp0.csettleorgid as csettleorgid,
         |temp0.crprofitcenterid as crprofitcenterid,
         |temp0.dsenddate as dsenddate,
         |temp0.dr as dr,
         |temp0.vsrccode as vsrccode,
         |temp0.nnum as nnum,
         |(case when j.name like '%荣盛新材%' or j.name like '%越友有限%' then nmny/3500 else nmny end) as nmny,
         |temp0.norigdiscount as norigdiscount,
         |temp0.nqtorigtaxprice as nqtorigtaxprice,
         |temp0.nqtorigtaxnetprc as nqtorigtaxnetprc,
         |temp0.nqtorignetprice as nqtorignetprice,
         |temp0.norigmny as norigmny,
         |temp0.norigtaxmny as norigtaxmny,
         |temp0.nqttaxnetprice as nqttaxnetprice,
         |temp0.nqtnetprice as nqtnetprice,
         |temp0.ntaxmny as ntaxmny,
         |temp0.nitemdiscountrate as nitemdiscountrate,
         |temp0.nqttaxprice as nqttaxprice,
         |temp0.ntaxrate as ntaxrate,
         |temp0.day as day
         |from temp0
         |LEFT JOIN dim.dim_org_salesorg_nc j
         |ON temp0.csaleorgid = j.pk_salesorg
         |),temp3 as (
         |--获取客户维度表数据
         |select  a.pk_org,a.pk_customer,a.pk_custsaleclass, b.name fq,c.name sjbm ,d.name tjsyb
         |,e.code pscode,e.name psnname,f.name qdlx,g.pk_salesorg,a.pk_financeorg
         |from dim.dim_bd_custsale_nc a
         |left join dim.dim_bd_custsaleclass_nc b on a.pk_custsaleclass=b.pk_custsaleclass
         |left join dim.dim_bd_custsaleclass_nc c on b.pk_father =c.pk_custsaleclass
         |left join dim.dim_bd_custsaleclass_nc d on substr(b.code,1,2)=d.code
         |left join dim.dim_bd_psndoc_nc e on a.respperson =e.pk_psndoc
         |left join dim.dim_bd_channeltype_nc f on a.channel=f.pk_type
         |left join dim.dim_org_salesorg_nc g on d.name=g.name
         |where a.pk_custsaleclass<>'~'
         |),temp4 as (
         |--获取物料维度表数据
         |select a.code as code,
         |a.name as name,
         |a.creationtime as creationtime,
         |CASE
         |WHEN a.enablestate = 1 THEN '未启用'
         |WHEN a.enablestate = 2 THEN '已启用'
         |WHEN a.enablestate = 3 THEN '已停用'
         |ELSE NULL -- 处理其他未指定的情况
         |END AS statue,
         |a.materialshortname as materialshortname,
         |a.materialspec as material_spec,
         |b.def1 as material_df1,
         |c.name as material_series,
         |e1.name as material_class1,
         |e2.name as material_class2,
         |e3.name as material_industry1,
         |e4.name as material_industry2,
         |e5.name as research_team,
         |g.name as group_name,
         |h.name as place_of_origin
         |from dim.dim_bd_material_v_nc a
         |left join dim.dim_bd_marbasclass_nc b on substr(a.code,1,3)=b.code
         |left join dim.dim_bd_marbasclass_nc c on a.pk_marbasclass=c.pk_marbasclass
         |left join dim.dim_bd_defdoc_nc e1 on  a.def13 = e1.pk_defdoc
         |left join dim.dim_bd_defdoc_nc e2 on  a.def11 = e2.pk_defdoc
         |left join dim.dim_bd_defdoc_nc e3 on  a.def12 = e3.pk_defdoc
         |left join dim.dim_bd_defdoc_nc e4 on  a.def1 = e4.pk_defdoc
         |left join dim.dim_bd_defdoc_nc e5 on  a.def5 = e5.pk_defdoc
         |left join dim.dim_org_group_nc g on  a.pk_group=g.pk_group
         |left join dim.dim_bd_defdoc_nc h on  a.prodarea =h.pk_defdoc
         |),temp5 as(
         |--主表跟子表合并
         |select
         |a.cdeliverybid as cdeliverybid,
         |a.cdeliveryid as cdeliveryid,
         |b.vbillcode as vbillcode,
         |a.vsrccode as vsrccode,
         |a.dbilldate as dbilldate,
         |a.dsenddate as dsenddate,
         |(CASE WHEN b.fstatusflag = '1' THEN '自由'
         |WHEN b.fstatusflag = '2' THEN '审批通过'
         |WHEN b.fstatusflag = '3' THEN '冻结'
         |WHEN b.fstatusflag = '4' THEN '关闭'
         |WHEN b.fstatusflag = '7' THEN '审批中'
         |WHEN b.fstatusflag = '8' THEN '审批不通过'
         |WHEN b.fstatusflag = '5' THEN '失效' ELSE '' END) AS vbillstatue,
         |c.code material_code,
         |c.name material_name,
         |c.materialspec material_spec,
         |d.name material_class,
         |p.def1 material_df1,
         |temp4.material_series as material_series,
         |temp4.material_class1 as material_class1,
         |temp4.material_class2 as material_class2,
         |temp4.material_industry1 as material_industry1,
         |temp4.material_industry2 as material_industry2,
         |temp4.research_team as research_team,
         |temp4.place_of_origin as place_of_origin,
         |e.code as customer_code,
         |e.name customer_name,
         |o.qdlx customer_type,
         |ii.code as invoice_customer_code,
         |ii.name as invoice_customer_name,
         |l.name as group_name,
         |q.name as org_financeorg_name,
         |f.name as org_salesorg_name,  --销售组织名称
         |m.name as org_trafficorg_name, --物流组织名称
         |o.tjsyb as tjsyb,          --事业部
         |o.sjbm as sjbm,            --部门组织名称
         |o.fq as fq,                --片区
         |o.pscode as pscode,        --业务员工号
         |o.psnname as psnname,      --业务员名称
         |nvl(a.nnum,0) as nnum,
         |nvl(a.nmny,0) as local_currency_delivery,
         |nvl(a.norigdiscount,0) as norigdiscount,
         |nvl(a.nqtorigtaxprice,0) as nqtorigtaxprice,
         |nvl(a.nqtorigtaxnetprc,0) as nqtorigtaxnetprc,
         |nvl(a.nqtorignetprice,0) as nqtorignetprice,
         |nvl(a.norigmny,0)  as norigmny,
         |nvl(a.norigtaxmny,0) as norigtaxmny,
         |nvl(a.nqttaxnetprice,0) as nqttaxnetprice,
         |nvl(a.nqtnetprice,0) as nqtnetprice,
         |nvl(a.ntaxmny,0) as ntaxmny,
         |nvl(CASE WHEN b.vdef4 = '1001A11000000000058S' THEN (a.norigdiscount - (1 - a.nitemdiscountrate / 100) * a.nnum * a.nqttaxprice) / (1 + a.ntaxrate / 100)
         |ELSE 0 END,0) as rebate_discount_amount,
         |nvl(a.nmny+(CASE WHEN b.vdef4 = '1001A11000000000058S' THEN (a.norigdiscount - (1 - a.nitemdiscountrate / 100) * a.nnum * a.nqttaxprice) / (1 + a.ntaxrate / 100)
         |ELSE 0 END),0) as nmny,
         |j.name as address,
         |substring(a.day,0,4) as year,
         |CASE
         |-- 当提取的月份是 01 - 03 时，判定为一季度
         |WHEN substring(a.day, 5, 2) IN ('01', '02', '03') THEN '1'
         |-- 当提取的月份是 04 - 06 时，判定为二季度
         |WHEN substring(a.day, 5, 2) IN ('04', '05', '06') THEN '2'
         |-- 当提取的月份是 07 - 09 时，判定为三季度
         |WHEN substring(a.day, 5, 2) IN ('07', '08', '09') THEN '3'
         |-- 当提取的月份是 10 - 12 时，判定为四季度
         |WHEN substring(a.day, 5, 2) IN ('10', '11', '12') THEN '4'
         |-- 若月份不在上述范围内，返回 NULL
         |ELSE NULL
         |END AS quarter,
         |substring(a.day,0,6) as month,
         |cast(substring(a.day, 5, 2) as int) as month2,
         |(floor((dayofmonth(from_unixtime(unix_timestamp(a.day, 'yyyyMMdd'), 'yyyy-MM-dd')) - 1) / 7) + 1) AS week,
         |a.day as day,
         |row_number() over(partition by a.cdeliverybid   order by a.dsenddate desc) as rn,
         |a.day as pt_date
         |FROM temp2 a  --发货头
         |LEFT JOIN temp1  b  ON a.cdeliveryid = b.cdeliveryid --发货体
         |LEFT JOIN dim.dim_bd_material_v_nc c ON a.cmaterialvid = c.pk_source --物料表
         |LEFT JOIN dim.dim_bd_marbasclass_nc d ON c.pk_marbasclass = d.pk_marbasclass --物料分类
         |LEFT JOIN dim.dim_bd_customer_nc e ON a.cordercustid = e.pk_customer --客户表
         |LEFT JOIN dim.dim_org_salesorg_nc f ON a.csaleorgid = f.pk_salesorg --销售组织
         |LEFT JOIN dim.dim_org_dept_nc g ON a.cdeptid = g.pk_dept --部门组织
         |LEFT JOIN dim.dim_bd_psndoc_nc h ON a.cemployeeid = h.pk_psndoc --人员信息
         |LEFT JOIN dim.dim_bd_customer_nc ii ON a.cinvoicecustid = ii.pk_customer --客户表
         |LEFT JOIN dim.dim_bd_stordoc_nc j ON a.csendstordocid = j.pk_stordoc --发货仓库
         |LEFT JOIN dim.dim_bd_supplier_nc k ON a.cvendorid = k.pk_supplier --供应商信息
         |LEFT JOIN dim.dim_org_group_nc l ON a.pk_group = l.pk_group --组织集体
         |LEFT JOIN dim.dim_org_trafficorg_nc m ON a.pk_org = m.pk_trafficorg --物料组织
         |LEFT JOIN temp3 o ON a.csaleorgid = o.pk_org   --片区
         |AND a.cordercustid = o.pk_customer
         |left join dim.dim_bd_marbasclass_nc p on substr (d.code, 1, 3) = p.code
         |left join dim.dim_org_financeorg_nc q on a.csettleorgid = q.pk_financeorg
         |left join temp4 on temp4.code=c.code
         |WHERE b.pk_group = '0001A110000000000EWV' and a.dr=0 and b.dr=0   --组织为回天集团
         |),temp6 as(
         |--合并后去重
         |select
         |cdeliverybid,
         |cdeliveryid,
         |vbillcode,
         |vsrccode,
         |dbilldate,
         |dsenddate,
         |vbillstatue,
         |material_code,
         |material_name,
         |material_spec,
         |material_class,
         |material_df1,
         |material_series,
         |material_class1,
         |material_class2,
         |material_industry1,
         |material_industry2,
         |research_team,
         |place_of_origin,
         |customer_code,
         |customer_name,
         |customer_type,
         |invoice_customer_code,
         |invoice_customer_name,
         |group_name,
         |org_financeorg_name,
         |org_salesorg_name,
         |org_trafficorg_name,
         |tjsyb,
         |sjbm,
         |fq,
         |pscode,
         |psnname,
         |nnum,
         |local_currency_delivery,
         |norigdiscount,
         |nqtorigtaxprice,
         |nqtorigtaxnetprc,
         |nqtorignetprice,
         |norigmny,
         |norigtaxmny,
         |nqttaxnetprice,
         |nqtnetprice,
         |ntaxmny,
         |rebate_discount_amount,
         |nmny,
         |address,
         |year,
         |quarter,
         |month,
         |month2,
         |week,
         |day,
         |pt_date
         |FROM temp5
         |where rn=1
         |),temp7 as(
         |select
         |day,
         |week
         |from dim.dim_time_in_week
         |)
         |INSERT OVERWRITE dwd.dwd_delivery_nc PARTITION (pt_date)
         |select
         |cdeliverybid,
         |cdeliveryid,
         |vbillcode,
         |vsrccode,
         |dbilldate,
         |dsenddate,
         |vbillstatue,
         |material_code,
         |material_name,
         |material_spec,
         |material_class,
         |material_df1,
         |material_series,
         |material_class1,
         |material_class2,
         |material_industry1,
         |material_industry2,
         |research_team,
         |place_of_origin,
         |customer_code,
         |customer_name,
         |customer_type,
         |invoice_customer_code,
         |invoice_customer_name,
         |group_name,
         |org_financeorg_name,
         |org_salesorg_name,
         |org_trafficorg_name,
         |tjsyb,
         |sjbm,
         |fq,
         |pscode,
         |psnname,
         |nnum,
         |local_currency_delivery,
         |norigdiscount,
         |nqtorigtaxprice,
         |nqtorigtaxnetprc,
         |nqtorignetprice,
         |norigmny,
         |norigtaxmny,
         |nqttaxnetprice,
         |nqtnetprice,
         |ntaxmny,
         |rebate_discount_amount,
         |(case when org_financeorg_name like '%荣盛新材%' or org_financeorg_name like '%越友有限%' then nmny/3500 else nmny end) as nmny,
         |address,
         |year,
         |quarter,
         |month,
         |month2,
         |nvl(temp7.week,temp6.week) as week,
         |temp6.day as day,
         |temp6.day as pt_date
         |FROM temp6 left join temp7
         |on temp6.day=temp7.day
  """.stripMargin


    log.info("数据插入到 dwd_delivery_nc 表")
    log.info("sql为："+text)
    spark.sql(text)






  }

  override def get_dws_delivery_in_salor_nc_w(spark: SparkSession, day: String, lastday: String, numPartition: Int, log: Logger): DataFrame= {
    log.info("day====" + day)
    val year = day.substring(0, 4)
    val month = day.substring(0, 6)
    val lastmonth = lastday.substring(0, 6)
    val month2 = day.substring(4, 6).toInt
    val text =
      s"""
         |with temp0 as(
         |--事业部维度去重,过滤"新能源背板事业部"
         |select
         |ts,
         |pk_org,
         |pk_customer,
         |pk_custsaleclass,
         |fq,
         |sjbm ,
         |tjsyb,
         |pscode,
         |psnname,
         |qdlx,
         |pk_salesorg,
         |pk_financeorg,
         |rn
         |from(
         |select
         |a.ts as ts,
         |a.pk_org as pk_org,
         |a.pk_customer as pk_customer,
         |a.pk_custsaleclass as pk_custsaleclass,
         |b.name fq,
         |c.name sjbm ,
         |d.name tjsyb,
         |e.code pscode,
         |e.name psnname,
         |f.name qdlx,
         |g.pk_salesorg as pk_salesorg,
         |a.pk_financeorg as pk_financeorg,
         |row_number() over (partition by d.name,e.code,e.name order by a.ts desc) as rn
         |from dim.dim_bd_custsale_nc a
         |left join dim.dim_bd_custsaleclass_nc b on a.pk_custsaleclass=b.pk_custsaleclass and b.enablestate =2
         |left join dim.dim_bd_custsaleclass_nc c on b.pk_father =c.pk_custsaleclass and c.enablestate =2
         |left join dim.dim_bd_custsaleclass_nc d on substr(b.code,1,2)=d.code and d.enablestate =2
         |left join dim.dim_bd_psndoc_nc e on a.respperson =e.pk_psndoc
         |left join dim.dim_bd_channeltype_nc f on a.channel=f.pk_type
         |left join dim.dim_org_salesorg_nc g on d.name=g.name
         |where a.pk_custsaleclass<>'~'
         |and d.name  in(
         |'汽化事业部'
         |,'电子事业部'
         |,'外贸东南亚事业部'
         |,'工业事业部'
         |,'锂电事业部'
         |,'微电子事业部'
         |,'包装事业部'
         |,'新能源硅胶事业部'
         |,'建筑胶事业部'
         |,'外贸中东事业部'
         |)
         |)t
         |where rn=1
         |),temp1 as (
         |--获取月度销售目标
         |select
         |a.jobno as jobno,
         |a.syb as syb,
         |a.cust_sale_type as cust_sale_type,
         |a.yearint as year,
         |a.monthint as month,
         |b.week as week,
         |a.target_amount as target_amount,
         |b.allocation_rate as allocation_rate,
         |(a.target_amount*b.allocation_rate)/100 as week_target_amount,
         |(a.target_amount_up*b.allocation_rate)/100 as week_target_amount_up
         |from dim.dim_crm_user_target a left join ods.ods_tjsyb_allocation_nc b
         |on a.yearint=b.year
         |and a.monthint=b.month
         |and a.syb=b.tjsyb
         |where a.yearint='${year}' and a.monthint='${month2}'
         |and a.syb!='常州公司'
         |),temp12 as(
         |--获取每周目标占比
         |select
         |temp1.syb as tjsyb,
         |temp0.sjbm as sjbm,
         |temp0.fq as fq,
         |temp1.jobno as pscode,
         |temp0.psnname as psnname,
         |temp1.cust_sale_type as cust_sale_type,
         |temp1.year as year,
         |temp1.month as month,
         |temp1.week as week,
         |temp1.target_amount as target_amount,
         |temp1.allocation_rate as allocation_rate,
         |temp1.week_target_amount as week_target_amount,
         |temp1.week_target_amount_up as week_target_amount_up
         |from temp1 left join temp0
         |on temp1.jobno=temp0.pscode
         |),temp2 as(
         |--获取发货单数据
         |select
         |tjsyb,
         |sjbm,
         |fq,
         |pscode,
         |psnname,
         |sum(num) as num,
         |sum(nmny) as nmny,
         |substring(month,0,4) as year,
         |month,
         |CAST(SUBSTRING(month, 5, 2) AS INT) AS month2,
         |week,
         |month as pt_month
         |from dwd.dwd_delivery_nc
         |where pt_date>='${lastday}'
         |and pt_date<='${day}'
         |group by tjsyb,sjbm,fq,pscode,psnname,month,week
         |),temp3 as(
         |--根据pscode,tjsyb,year,month2,week关联合并发货单数据核部门每周的销售目标
         |--员工所在的事业部，员工编号，时间进行关联
         |select
         |nvl(temp12.tjsyb,'null') as tjsyb,
         |nvl(temp12.sjbm,'null') as sjbm,
         |nvl(temp12.fq,'null') as fq,
         |nvl(temp12.pscode,'null') as pscode,
         |nvl(temp12.psnname,'null') as psnname,
         |nvl(temp2.num,0.0) as num,
         |nvl(temp2.nmny,0.0) as nmny,
         |temp12.year as year,
         |temp12.month as month,
         |temp12.week as week,
         |LAG(temp2.nmny, 1,0.0) OVER (PARTITION BY temp2.pscode,temp2.year,temp2.month ORDER BY temp2.year,temp2.month,temp2.week asc) AS last_nmny,
         |nvl(temp12.week_target_amount,0.0) as week_target_amount,
         |nvl(temp12.week_target_amount_up,0.0) as week_target_amount_up
         |from temp12 left join temp2
         |on temp2.pscode=temp12.pscode
         |and temp12.tjsyb=temp2.tjsyb
         |and temp12.sjbm=temp2.sjbm
         |and temp2.year=temp12.year
         |and temp2.month2=temp12.month
         |and temp2.week=temp12.week
         |order by temp2.year desc
         |),temp4 as(
         |--数据去重，并生成id，准备spark数据计算
         |select
         |concat(tjsyb,'#',sjbm,'#',fq,'#',pscode,'#',psnname,'#',year,'#',month) as id,
         |tjsyb,
         |sjbm,
         |fq,
         |pscode,
         |psnname,
         |year,
         |month,
         |week,
         |num,
         |nmny,
         |nvl(last_nmny,0.0) as last_nmny,
         |week_target_amount,
         |week_target_amount_up,
         |row_number() over (partition by tjsyb,sjbm,fq,pscode,psnname,year,month,week order by month,week asc) as rn
         |from temp3
         |)
         |select
         |id,
         |tjsyb,
         |sjbm,
         |fq,
         |pscode,
         |psnname,
         |year,
         |month,
         |'' as month2,
         |week,
         |num,
         |nmny,
         |last_nmny,
         |week_target_amount,
         |week_target_amount_up
         |from temp4
         |where rn=1
  """.stripMargin
    import spark.implicits._
    log.info("sql====" + text)
    val targetDs: Dataset[TargetWeek] = spark.sql(text).as[TargetWeek]
    val order_Original_Data = targetDs.rdd.groupBy(_.id).mapValues(data => {
      // 默认为正序排列，要想倒序： data.toArray.sortBy( _.gpss_time).reverse
      data.toArray.sortBy(_.week)
    })

    val target: Dataset[TargetWeekResult] = order_Original_Data.flatMap(line => {

      val arr: Array[TargetWeek] = line._2
      val sessionResult: ArrayBuffer[TargetWeekResult] = new ArrayBuffer[TargetWeekResult]()

      var targer_week1 = 0.0
      var targer_week1_up = 0.0
      var targer_week2 = 0.0
      var targer_week2_up = 0.0
      var targer_week3 = 0.0
      var targer_week3_up = 0.0
      var targer_week4 = 0.0
      var targer_week4_up = 0.0
      var targer_week5 = 0.0
      var targer_week5_up = 0.0
      var diff=0.0
      var diff_up=0.0
      arr.foreach(data => {
        val tjsyb = data.tjsyb
        val sjbm = data.sjbm
        val fq = data.fq
        val pscode = data.pscode
        val psnname = data.psnname
        val year = data.year
        val monthint = data.month
        val formattedMonth = if (monthint.toInt < 10) s"0$monthint" else monthint.toString
        val month = s"$year$formattedMonth"
        val week = data.week
        val num = data.num
        val nmny = data.nmny
        val last_nmny = data.last_nmny
        val week_target_amount = data.week_target_amount
        val week_target_amount_up = data.week_target_amount_up
        val result = new TargetWeekResult
        result.tjsyb = tjsyb
        result.sjbm = sjbm
        result.fq = fq
        result.pscode = pscode
        result.psnname = psnname
        result.year = year
        result.month = month
        result.month2 = monthint
        result.week = week
        result.num = num
        result.nmny = nmny
        result.last_nmny = last_nmny
        result.week_target_amount = week_target_amount
        result.week_target_amount_up = week_target_amount_up

        if (week.contains("1")) {
          targer_week1 = week_target_amount
          targer_week1_up = week_target_amount_up
          result.target = targer_week1
          result.target_up = targer_week1_up
          result.diff = 0.0
        } else if (week.contains("2")) {
          diff=(targer_week1 - last_nmny)
          diff_up=(targer_week1_up - last_nmny)
          result.diff = diff
          result.diff_up = diff_up
          if (targer_week1 - last_nmny > 0) {
            targer_week2 = week_target_amount +diff
          } else {
            targer_week2 = week_target_amount
          }

          if (targer_week1_up - last_nmny > 0) {
            targer_week2_up = week_target_amount_up +diff_up
          } else {
            targer_week2_up = week_target_amount_up
          }

          result.target = targer_week2
          result.target_up = targer_week2_up
        } else if (week.contains("3")) {
          diff=(targer_week2 - last_nmny)
          diff_up=(targer_week2_up - last_nmny)
          result.diff = diff
          result.diff_up = diff_up
          if (targer_week2 - last_nmny > 0) {
            targer_week3 = week_target_amount + (targer_week2 - last_nmny)
          } else {
            targer_week3 = week_target_amount
          }

          if (targer_week2_up - last_nmny > 0) {
            targer_week3_up = week_target_amount_up + (targer_week2_up - last_nmny)
          } else {
            targer_week3_up = week_target_amount_up
          }
          result.target = targer_week3
          result.target_up = targer_week3_up

        } else if (week.contains("4")) {
          diff=(targer_week3 - last_nmny)
          diff_up=(targer_week3_up - last_nmny)
          result.diff = diff
          result.diff_up = diff_up
          if (targer_week3 - last_nmny > 0) {
            targer_week4 = week_target_amount +diff
          } else {
            targer_week4 = week_target_amount
          }

          if (targer_week3_up - last_nmny > 0) {
            targer_week4_up = week_target_amount_up +diff_up
          } else {
            targer_week4_up = week_target_amount_up
          }
          result.target = targer_week4
          result.target_up= targer_week4_up

        } else if (week.contains("5")) {
          diff=(targer_week4 - last_nmny)
          diff_up=(targer_week4_up - last_nmny)
          result.diff = diff
          result.diff_up = diff_up
          if (targer_week4 - last_nmny > 0) {
            targer_week5 = week_target_amount + diff
          } else {
            targer_week5 = week_target_amount
          }
          // up的处理
          if (targer_week4_up - last_nmny > 0) {
            targer_week5_up = week_target_amount_up + diff_up
          } else {
            targer_week5_up = week_target_amount_up
          }
          result.target=targer_week5
          result.target_up=targer_week5_up
        }
        sessionResult.append(result)

      })
      sessionResult
    }).toDF().as[TargetWeekResult]

    //数据插入到hive表
    log.info("数据插入到hive表中。。。。。")
    target.createTempView("tt")
    val sql =
      s"""
         |insert overwrite table dws.dws_delivery_in_salor_nc_w PARTITION (pt_month)
         |select
         |tjsyb,
         |sjbm,
         |fq,
         |pscode,
         |psnname,
         |num,
         |nmny,
         |year,
         |month,
         |month2,
         |week,
         |last_nmny,
         |week_target_amount,
         |week_target_amount_up,
         |target,
         |target_up,
         |diff,
         |diff_up,
         |month as pt_month
         |from(
         |select
         |tjsyb,
         |sjbm,
         |fq,
         |pscode,
         |psnname,
         |num,
         |nmny,
         |year,
         |month,
         |month2,
         |week,
         |last_nmny,
         |week_target_amount,
         |week_target_amount_up,
         |target,
         |target_up,
         |diff,
         |diff_up,
         |row_number() over (partition by tjsyb,sjbm,fq,pscode,psnname,year,month,week order by month,week asc) as rn,
         |month as pt_month
         |from tt
         |)t
         |where t.rn=1
  """.stripMargin
    spark.sql(sql)

    val sql2 =
      s"""
          select
          tjsyb,
          sjbm,
          fq,
          pscode,
          psnname,
          num,
          nmny,
          year,
          month,
          month2,
          week,
          last_nmny,
          week_target_amount,
          week_target_amount_up,
          target,
          target_up,
          diff,
          diff_up,
          current_timestamp() as sysdate
          from(
          select
          tjsyb,
          sjbm,
          fq,
          pscode,
          psnname,
          num,
          nmny,
          year,
          month,
          month2,
          week,
          last_nmny,
          week_target_amount,
          week_target_amount_up,
          target,
          target_up,
          diff,
          diff_up,
          row_number() over (partition by tjsyb,sjbm,fq,pscode,psnname,year,month,week order by month,week asc) as rn,
          month as pt_month
          from tt
          )t
          where t.rn=1
  """.stripMargin
    val frame = spark.sql(sql2)
    frame
  }

  override def get_dwd_saleinvoice_nc(spark: SparkSession, day: String, lastday: String, numPartition: Int, log: Logger): DataFrame = {
    log.info("day====" + day)
    val month = day.substring(0, 6)
    val lastmonth = lastday.substring(0, 6)
    val text =
      s"""
          select
          csaleinvoiceid
          ,csaleorgid
          ,csaleinvoicebid
          ,vbillcode
          ,vsrccode
          ,dbilldate
          ,customer_code
          ,customer_name
          ,invoice_code
          ,invoice_name
          ,customer_type
          ,org_salesorg_name
          ,org_financeorg_name
          ,tjsyb
          ,sjbm
          ,storehouse
          ,fq
          ,pscode
          ,psnname
          ,research_team
          ,material_industry
          ,research_institute
          ,material_code
          ,material_name
          ,material_spec
          ,material_df1
          ,nnum
          ,norigtaxnetprice
          ,norigmny
          ,ntax
          ,norigtaxmny
          ,ntaxmny_price
          ,ntaxmny
          ,price_no_tax_currency
          ,nmny
          ,ntotalincomenum
          ,ntotalincomemny
          ,ntotalpaymny
          ,ntotalreceiv
          ,year
          ,quarter
          ,month
          ,month2
          ,week
          ,day
          ,current_timestamp() as sysdate
          from dwd.dwd_saleinvoice_nc
          where pt_date<='${day}'
          and pt_date>='${lastday}'
  """.stripMargin

    log.info("sql====" + text)
    val df: DataFrame = spark.sql(text)
    df


  }


  override def get_dws_saleinvoice_in_customer_nc_m(spark: SparkSession, day: String, lastday: String, numPartitions: Int, log: Logger): DataFrame = {
    log.info("day====" + day)
    val month = day.substring(0, 6)
    val lastmonth = lastday.substring(0, 6)
    val text =
      s"""
          select
          tjsyb
          ,sjbm
          ,psnname
          ,customer_code
          ,customer_name
          ,customer_type
          ,qdlx
          ,nnum
          ,nmny
          ,prev_month_num
           ,prev_month_nmny
           ,prev_year_num
           ,prev_year_nmny
          ,num_year_growth
          ,num_month_growth
          ,num_year_growth_rate
          ,num_month_growth_rate
          ,nmny_year_growth
          ,nmny_month_growth
          ,nmny_year_growth_rate
          ,nmny_month_growth_rate
          ,norigtaxnetprice
          ,norigmny
          ,ntax
          ,norigtaxmny
          ,ntaxmny_price
          ,ntaxmny
          ,price_no_tax_currency
          ,ntotalincomenum
          ,ntotalincomemny
          ,ntotalpaymny
          ,ntotalreceiv
          ,year
          ,quarter
          ,month
          ,month2
          ,current_timestamp() as sysdate
          from dws.dws_saleinvoice_in_customer_nc_m
          where pt_month<='${month}'
          and pt_month>='${lastmonth}'
  """.stripMargin

    log.info("sql====" + text)
    val df: DataFrame = spark.sql(text)
    df


  }


  override def get_dws_saleinvoice_in_customer_nc_d(spark: SparkSession, day: String, lastday: String, numPartitions: Int, log: Logger): DataFrame = {
    log.info("day====" + day)
    val text =
      s"""
           select
           tjsyb
           ,sjbm
           ,psnname
           ,customer_code
           ,customer_name
           ,customer_type
           ,material_df1
           ,nnum
           ,nmny
           ,norigtaxnetprice
           ,norigmny
           ,ntax
           ,norigtaxmny
           ,ntaxmny_price
           ,ntaxmny
           ,price_no_tax_currency
           ,ntotalincomenum
           ,ntotalincomemny
           ,ntotalpaymny
           ,ntotalreceiv
           ,year
           ,quarter
           ,month
           ,month2
           ,week
           ,day
           ,current_timestamp() as sysdate
           from dws.dws_saleinvoice_in_customer_nc_d
           where pt_date<='${day}'
           and pt_date>='${lastday}'
  """.stripMargin

    log.info("sql====" + text)
    val df: DataFrame = spark.sql(text)
    df


  }

  override def get_dws_saleinvoice_in_industry_nc_m(spark: SparkSession, day: String, lastday: String, numPartitions: Int, log: Logger): DataFrame = {
    //pt_date='${day}'
    log.info("day====" + day)
    val month = day.substring(0, 6)
    val lastmonth = lastday.substring(0, 6)
    val text =
      s"""
          select
          tjsyb
          ,material_class
          ,material_industry
          ,nnum
          ,nmny
          ,prev_month_num
          ,prev_month_nmny
          ,prev_year_num
          ,prev_year_nmny
          ,num_year_growth
          ,num_month_growth
          ,num_year_growth_rate
          ,num_month_growth_rate
          ,nmny_year_growth
          ,nmny_month_growth
          ,nmny_year_growth_rate
          ,nmny_month_growth_rate
          ,norigtaxnetprice
          ,norigmny
          ,ntax
          ,norigtaxmny
          ,ntaxmny_price
          ,ntaxmny
          ,price_no_tax_currency
          ,ntotalincomenum
          ,ntotalincomemny
          ,ntotalpaymny
          ,ntotalreceiv
          ,year
          ,quarter
          ,month
          ,month2
          ,current_timestamp() as sysdate
          from dws.dws_saleinvoice_in_industry_nc_m
          where pt_month<='${month}'
          and pt_month>='${lastmonth}'
  """.stripMargin

    log.info("sql====" + text)
    val df: DataFrame = spark.sql(text)
    df


  }


  override def get_dws_saleinvoice_in_industry_nc_d(spark: SparkSession, day: String, lastday: String, numPartitions: Int, log: Logger): DataFrame = {
    //pt_date='${day}'
    log.info("day====" + day)
    val text =
      s"""
          select
          tjsyb
          ,material_df1
          ,material_series
          ,material_class1
          ,material_class2
          ,material_industry1
          ,material_industry2
          ,num
          ,nmny
          ,norigtaxnetprice
          ,norigmny
          ,ntax
          ,norigtaxmny
          ,ntaxmny_price
          ,ntaxmny
          ,price_no_tax_currency
          ,ntotalincomenum
          ,ntotalincomemny
          ,ntotalpaymny
          ,ntotalreceiv
          ,year
          ,quarter
          ,month
          ,month2
          ,week
          ,day
          ,current_timestamp() as sysdate
          from dws.dws_saleinvoice_in_industry_nc_d
          where pt_date<='${day}'
          and pt_date>='${lastday}'
  """.stripMargin

    log.info("sql====" + text)
    val df: DataFrame = spark.sql(text)
    df


  }

  override def get_dws_saleinvoice_in_salor_nc_m(spark: SparkSession, day: String, lastday: String, numPartitions: Int, log: Logger): DataFrame = {

    //pt_date='${day}'
    log.info("day====" + day)
    val month = day.substring(0, 6)
    val lastmonth = lastday.substring(0, 6)
    val text =
      s"""
          select
          tjsyb,
          sjbm,
          fq,
          psncode,
          psnname,
          nnum,
          nmny,
          prev_year_num ,
          prev_month_num ,
          prev_year_nmny,
          prev_month_nmny,
          num_year_growth,
          num_month_growth,
          num_year_growth_rate,
          num_month_growth_rate,
          nmny_year_growth,
          nmny_month_growth,
          nmny_year_growth_rate,
          nmny_month_growth_rate,
          norigtaxnetprice,
          norigmny,
          ntax,
          norigtaxmny,
          ntaxmny_price,
          ntaxmny,
          price_no_tax_currency,
          ntotalincomenum,
          ntotalincomemny,
          ntotalpaymny,
          ntotalreceiv,
          year,
          quarter,
          month,
          month2,
          current_timestamp() as sysdate
          from dws.dws_saleinvoice_in_salor_nc_m
          where pt_month<='${month}'
          and pt_month>='${lastmonth}'
  """.stripMargin

    log.info("sql====" + text)
    val df: DataFrame = spark.sql(text)
    df
  }


  override def get_dws_saleinvoice_in_salor_nc_d(spark: SparkSession, day: String, lastday: String, numPartitions: Int, log: Logger): DataFrame = {
    //pt_date='${day}'
    log.info("day====" + day)
    val month = day.substring(0, 6)
    val lastmonth = lastday.substring(0, 6)
    val text =
      s"""
         |select
         |tjsyb
         |,sjbm
         |,fq
         |,psncode
         |,psnname
         |,nnum
         |,norigtaxnetprice
         |,norigmny
         |,ntax
         |,norigtaxmny
         |,ntaxmny_price
         |,ntaxmny
         |,price_no_tax_currency
         |,nmny
         |,ntotalincomenum
         |,ntotalincomemny
         |,ntotalpaymny
         |,ntotalreceiv
         |,year
         |,month
         |,month2
         |,quarter
         |,day
         |,current_timestamp() as sysdate
         |from dws.dws_saleinvoice_in_salor_nc_d
         |where pt_date<='${day}'
         |and pt_date>='${lastday}'
         |
  """.stripMargin

    log.info("sql====" + text)
    val df: DataFrame = spark.sql(text)
    df


  }

  override def get_dws_delivery_in_industry_nc_d(spark: SparkSession, day: String, lastday: String, numPartition: Int, log: Logger): DataFrame = {
    //pt_date='${day}'
    log.info("day====" + day)
    val text =
      s"""
          select
          tjsyb
          ,material_class
          ,material_df1
          ,material_series
          ,material_class1
          ,material_class2
          ,material_Industry1
          ,material_Industry2
          ,nnum
          ,nmny
          ,norigdiscount
          ,nqtorigtaxprice
          ,nqtorigtaxnetprc
          ,nqtorignetprice
          ,norigmny
          ,norigtaxmny
          ,nqttaxnetprice
          ,nqtnetprice
          ,ntaxmny
          ,rebate_discount_amount
          ,year
          ,quarter
          ,month
          ,day
          ,current_timestamp() as sysdate
          from dws.dws_delivery_in_industry_nc_d
          where pt_date<='${day}'
          and pt_date>='${lastday}'
  """.stripMargin

    log.info("sql====" + text)
    val df: DataFrame = spark.sql(text)


    df

  }


  override def get_dws_delivery_in_industry_nc_m(spark: SparkSession, day: String, lastday: String, numPartition: Int, log: Logger): DataFrame = {
    //pt_date='${day}'
    log.info("day====" + day)
    val month = day.substring(0, 6)
    val lastmonth = lastday.substring(0, 6)
    val text =
      s"""
         select
         tjsyb
         ,material_class
         ,material_industry
         ,nnum
         ,nmny
         ,prev_month_num
         ,prev_month_nmny
         ,prev_year_num
         ,prev_year_nmny
         ,num_year_growth
         ,num_month_growth
         ,num_year_growth_rate
         ,num_month_growth_rate
         ,nmny_year_growth
         ,nmny_month_growth
         ,nmny_year_growth_rate
         ,nmny_month_growth_rate
         ,norigdiscount
         ,nqtorigtaxprice
         ,nqtorigtaxnetprc
         ,nqtorignetprice
         ,norigmny
         ,norigtaxmny
         ,nqttaxnetprice
         ,nqtnetprice
         ,ntaxmny
         ,rebate_discount_amount
         ,year
         ,quarter
         ,month
         ,month2
         ,current_timestamp() as sysdate
         from dws.dws_delivery_in_industry_nc_m
         where pt_month<='${month}'
         and pt_month>='${lastmonth}'
  """.stripMargin

    log.info("sql====" + text)
    val df: DataFrame = spark.sql(text)
    df

  }

  //get_dwd_saleinvoice_custome_industry_nc_d
  override def get_dwd_saleinvoice_custome_industry_nc_d(spark: SparkSession, day: String, lastday: String, numPartitions: Int, log: Logger): DataFrame = {
    //pt_date='${day}'
    log.info("day====" + day)
    val month = day.substring(0, 6)
    val lastmonth = lastday.substring(0, 6)
    val text =
      s"""
         |select
         |vbillcode
         |,vsrccode
         |,dbilldate
         |,custome_code
         |,custome_name
         |,vbill_custome_code
         |,vbill_custome_name
         |,custome_type
         |,org_salesorg_name
         |,org_financeorg_name
         |,tjsyb
         |,sjbm
         |,storehouse
         |,fq
         |,psncode
         |,psnname
         |,qdlx
         |,research_team
         |,material_industry
         |,research_institute
         |,material_code
         |,material_name
         |,material_spec
         |,material_df1
         |,material_series
         |,material_class1
         |,material_class2
         |,nnum
         |,norigtaxnetprice
         |,norigmny
         |,ntax
         |,norigtaxmny
         |,ntaxmny_price
         |,ntaxmny
         |,price_no_tax_currency
         |,tax_free_nmny
         |,tax_free_rebate
         |,local_unit_cost
         |,total_cost_in_local_currency
         |,gross_profit_local
         |,nmny
         |,ntotalincomenum
         |,ntotalincomemny
         |,ntotalpaymny
         |,ntotalreceiv
         |,year
         |,quarter
         |,month
         |,month2
         |,day
         |,current_timestamp() as sysdate
         |from dwd.dwd_saleinvoice_custome_industry_nc_d
         |where pt_date<='${day}'
         |and pt_date>='${lastday}'
  """.stripMargin

    log.info("sql====" + text)
    val df: DataFrame = spark.sql(text)
    df


  }

  override def get_dws_tab_fhinfo_hb_nc_m(spark: SparkSession, day: String, lastday: String, numPartitions: Int, log: Logger): DataFrame = {

    //pt_date='${day}'
    log.info("day====" + day)
    val month = day.substring(0, 6)
    val lastmonth = lastday.substring(0, 6)
    val text =
      s"""
         select
         customer_code
         ,customer_name
         ,custshortname
         ,ysaleorg
         ,saleorgcode
         ,saleorgname
         ,financeorgcode
         ,financeorgname
         ,deptcode
         ,deptname
         ,psndoccode
         ,psndocname
         ,areacode
         ,areaname
         ,channeltypecode
         ,channeltypename
         ,linkman
         ,telphone
         ,currtype
         ,ys
         ,hk
         ,ye
         ,year
         ,quarter
         ,month
         ,month2
         ,current_timestamp() as sysdate
         from dws.dws_tab_fhinfo_hb_nc_m
         order by customer_code asc
  """.stripMargin

    log.info("sql====" + text)
    val df: DataFrame = spark.sql(text)
    df


  }

  override def get_dwd_delivery_nc(spark: SparkSession, day: String, lastday: String, numPartition: Int, log: Logger): DataFrame = {
    //pt_date='${day}'
    log.info("day====" + day)
    val text =
      s"""
         select
         cdeliverybid as cdeliverybid
         ,cdeliveryid as cdeliveryid
         ,vbillcode
         ,vsrccode
         ,dbilldate
         ,dsenddate
         ,vbillstatue
         ,material_code
         ,material_name
         ,material_spec
         ,material_class
         ,material_df1
         ,material_series
         ,material_class1
         ,material_class2
         ,material_industry1
         ,material_Industry2
         ,research_team
         ,place_of_origin
         ,customer_code
         ,customer_name
         ,customer_type
         ,invoice_customer_code
         ,invoice_customer_name
         ,group_name
         ,org_financeorg_name
         ,org_salesorg_name
         ,org_trafficorg_name
         ,tjsyb
         ,sjbm
         ,fq
         ,pscode
         ,psnname
         ,num
         ,local_currency_delivery
         ,norigdiscount
         ,nqtorigtaxprice
         ,nqtorigtaxnetprc
         ,nqtorignetprice
         ,norigmny
         ,norigtaxmny
         ,nqttaxnetprice
         ,nqtnetprice
         ,ntaxmny
         ,rebate_discount_amount
         ,nmny
         ,address
         ,year
         ,quarter
         ,month
         ,month2
         ,week
         ,day
         ,current_timestamp() as sysdate
         from dwd.dwd_delivery_nc
         where pt_date<='${day}'
         and pt_date>='${lastday}'
  """.stripMargin

    log.info("sql====" + text)
    val df: DataFrame = spark.sql(text)
    df
  }


  override def get_dws_delivery_in_salor_nc_m(spark: SparkSession, day: String, lastday: String, numPartition: Int, log: Logger): DataFrame = {

    //pt_date='${day}'
    log.info("day====" + day)
    val month = day.substring(0, 6)
    val lastmonth = lastday.substring(0, 6)
    val text =
      s"""
         |select
         |tjsyb
         |,sjbm
         |,fq
         |,pscode
         |,psnname
         |,num
         |,nmny
         |,prev_month_num
         |,prev_month_nmny
         |,prev_year_num
         |,prev_year_nmny
         |,num_year_growth
         |,num_month_growth
         |,num_year_growth_rate
         |,num_month_growth_rate
         |,nmny_year_growth
         |,nmny_month_growth
         |,nmny_year_growth_rate
         |,nmny_month_growth_rate
         |,norigdiscount
         |,nqtorigtaxprice
         |,nqtorigtaxnetprc
         |,nqtorignetprice
         |,norigmny
         |,norigtaxmny
         |,nqttaxnetprice
         |,nqtnetprice
         |,ntaxmny
         |,rebate_discount_amount
         |,local_currency_delivery
         |,total_deliver_order
         |,total_deliver_of_freedom
         |,total_deliver_of_approve
         |,total_deliver_of_block
         |,total_deliver_of_close
         |,total_deliver_of_approval
         |,total_deliver_of_approval_denied
         |,total_deliver_of_expire
         |,year
         |,quarter
         |,month
         |,month2
         |,current_timestamp() as sysdate
         |from dws.dws_delivery_in_salor_nc_m
         |where pt_month<='${month}'
         |and pt_month>='${lastmonth}'
  """.stripMargin

    log.info("sql====" + text)
    val df: DataFrame = spark.sql(text)
    df


  }

  /**
   * 获取Tos上的数据
   *
   * @param spark
   * @param day ：日期
   * @param numPartion
   * @return
   */

  override def get_dws_delivery_in_salor_nc_d(spark: SparkSession, day: String, lastday: String, numPartion: Int, log: Logger): DataFrame = {

    import spark.implicits._

    //pt_date='${day}'
    log.info("day====" + day)
    val text =
      s"""
         select
         tjsyb,
         sjbm,
         fq,
         pscode,
         psnname,
         num,
         nmny,
         norigdiscount,
         nqtorigtaxprice,
         nqtorigtaxnetprc,
         nqtorignetprice,
         norigmny,
         norigtaxmny,
         nqttaxnetprice,
         nqtnetprice,
         ntaxmny,
         rebate_discount_amount,
         local_currency_delivery,
         total_deliver_order,
         total_deliver_of_freedom,
         total_deliver_of_approve,
         total_deliver_of_block,
         total_deliver_of_close,
         total_deliver_of_approval,
         total_deliver_of_approval_denied,
         total_deliver_of_expire,
         year,
         quarter,
         month,
         month2,
         week,
         day,
         current_timestamp() as sysdate
         from dws.dws_delivery_in_salor_nc_d
         where pt_date<='${day}'
         and pt_date>='${lastday}'
  """.stripMargin

    val df: DataFrame = spark.sql(text)
    df

  }


  /**
   * 将日期转换成时间戳
   *
   * @param day 日期
   * @return 时间戳
   */
  def getTime(day: String): Long = {
    new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(day).getTime
  }


  /**
   * 计算经度纬度
   *
   * @param deg 度
   * @param min 分
   * @param sec 秒
   * @return 度+(分/60)+秒/3600
   */
  def getLatOrLong(deg: String, min: String, sec: String, index: Int): Option[Double] = {
    var itemValue: Double = 0.0
    try {
      if (deg != null && deg.contains(",") && min != null && min.contains(",") && sec != null && sec.contains(",")) {
        val degValues = deg.split(",")
        val minValues = min.split(",")
        val secValues = sec.split(",")

        var degNum = 0.0
        var minNum = 0.0
        var secNUM = 0.0

        if (index <= degValues.length) {
          degNum = degValues(index).toDouble
        }

        if (index <= minValues.length) {
          minNum = minValues(index).toDouble
        }

        if (index <= secValues.length) {
          secNUM = secValues(index).toDouble
        }

        itemValue = (degNum + minNum / 60 + secNUM / 3600)

      }
    } catch {
      case e: Exception =>
        println(e, "error getLatOrLong deg=" + deg + " min=" + min + " sec=" + sec + " index=" + index)
    }

    Some(itemValue)
  }


  override def getNextMonth(day: String): String = {
    val monthStar = day.substring(0, 7) + "-01 00:00:00"
    val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
    val dt = sdf.parse(monthStar)
    val rightNow = Calendar.getInstance
    rightNow.setTime(dt)
    rightNow.add(Calendar.MONTH, 1)
    val dt1 = rightNow.getTime
    val reStr = sdf.format(dt1)
    /*    val format = DateTimeFormatter.ofPattern( "yyyy-MM-dd HH:mm:ss" )
        val format2 = DateTimeFormatter.ofPattern( "yyyyMMdd" )
        val localDate = LocalDate.parse( monthStar, format )
        val str = localDate.plusDays( 30 ).format( format2 )*/

    reStr
  }

  override def getNextDay(day: String): String = {
    val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
    val dt = sdf.parse(day)
    val rightNow = Calendar.getInstance
    rightNow.setTime(dt)
    rightNow.add(Calendar.DATE, 1)
    val dt1 = rightNow.getTime
    val reStr = sdf.format(dt1)
    reStr
  }


  override def write2dws_saleinvoice_in_customer_nc_m(spark: SparkSession, day: String, lastday: String, ds: DataFrame, numPartition: Int, log: Logger): Unit = {

    import spark.implicits._
    log.info("数据写入到mysql 中。。。")
    log.info("删除mysql里边的数据")
    val month = day.substring(0, 6)
    val lastmonth = lastday.substring(0, 6)
    MysqlUtils.delete_dws_saleinvoice_in_customer_nc_m(month, lastmonth);

    val mysqlConf = new MysqlConfiguration

    log.info("数据写入到mysql")
    ds.write.mode("append").format("jdbc").option("url", mysqlConf.mysqlJdbcUrl)
      .option("dbtable", "dws_saleinvoice_in_customer_nc_m")
      .option("user", mysqlConf.dataSourceUser)
      .option("password", mysqlConf.dataSourcePassword)
      .option("driver", mysqlConf.mysqlJdbcDriver)
      .option("batchsize", 10000)
      .save()


  }


  override def write2dws_saleinvoice_in_industry_nc_m(spark: SparkSession, day: String, lastday: String, ds: DataFrame, numPartition: Int, log: Logger): Unit = {
    import spark.implicits._
    log.info("数据写入到mysql 中。。。")
    log.info("删除mysql里边的数据")
    val month = day.substring(0, 6)
    val lastmonth = lastday.substring(0, 6)
    MysqlUtils.delete_dws_saleinvoice_in_industry_nc_m(month, lastmonth);

    val mysqlConf = new MysqlConfiguration

    log.info("数据写入到mysql")
    ds.write.mode("append").format("jdbc").option("url", mysqlConf.mysqlJdbcUrl)
      .option("dbtable", "dws_saleinvoice_in_industry_nc_m")
      .option("user", mysqlConf.dataSourceUser)
      .option("password", mysqlConf.dataSourcePassword)
      .option("driver", mysqlConf.mysqlJdbcDriver)
      .option("batchsize", 10000)
      .save()


  }


  override def write2dws_saleinvoice_in_industry_nc_d(spark: SparkSession, day: String, lastday: String, ds: DataFrame, numPartition: Int, log: Logger): Unit = {
    import spark.implicits._
    log.info("数据写入到mysql 中。。。")
    log.info("删除mysql里边的数据")
    MysqlUtils.delete_dws_saleinvoice_in_industry_nc_d(day, lastday);

    val mysqlConf = new MysqlConfiguration

    log.info("数据写入到mysql")
    ds.write.mode("append").format("jdbc").option("url", mysqlConf.mysqlJdbcUrl)
      .option("dbtable", "dws_saleinvoice_in_industry_nc_d")
      .option("user", mysqlConf.dataSourceUser)
      .option("password", mysqlConf.dataSourcePassword)
      .option("driver", mysqlConf.mysqlJdbcDriver)
      .option("batchsize", 10000)
      .save()


  }


  override def write2dws_delivery_in_salor_nc_m(spark: SparkSession, day: String, lastday: String, ds: DataFrame, numPartition: Int, log: Logger): Unit = {
    import spark.implicits._
    log.info("数据写入到mysql 中。。。")
    log.info("删除mysql里边的数据")
    val month = day.substring(0, 6)
    val lastmonth = lastday.substring(0, 6)
    MysqlUtils.delete_dws_delivery_in_salor_nc_m(month, lastmonth);

    val mysqlConf = new MysqlConfiguration

    log.info("数据写入到mysql")
    ds.write.mode("append").format("jdbc").option("url", mysqlConf.mysqlJdbcUrl)
      .option("dbtable", "dws_delivery_in_salor_nc_m")
      .option("user", mysqlConf.dataSourceUser)
      .option("password", mysqlConf.dataSourcePassword)
      .option("driver", mysqlConf.mysqlJdbcDriver)
      .option("batchsize", 10000)
      .save()


  }


  override def write2dws_saleinvoice_in_salor_nc_m(spark: SparkSession, day: String, lastday: String, ds: DataFrame, numPartition: Int, log: Logger): Unit = {
    import spark.implicits._
    log.info("数据写入到mysql 中。。。")
    log.info("删除mysql里边的数据")
    val month = day.substring(0, 6)
    val lastmonth = lastday.substring(0, 6)
    MysqlUtils.delete_dws_saleinvoice_in_salor_nc_m(month, lastmonth);

    val mysqlConf = new MysqlConfiguration

    log.info("数据写入到mysql")
    ds.write.mode("append").format("jdbc").option("url", mysqlConf.mysqlJdbcUrl)
      .option("dbtable", "dws_saleinvoice_in_salor_nc_m")
      .option("user", mysqlConf.dataSourceUser)
      .option("password", mysqlConf.dataSourcePassword)
      .option("driver", mysqlConf.mysqlJdbcDriver)
      .option("batchsize", 10000)
      .save()


  }


  override def write2dws_delivery_in_salor_nc_d(spark: SparkSession, day: String, lastday: String, ds: DataFrame, numPartion: Int, log: Logger): Unit = {

    import spark.implicits._
    log.info("数据写入到mysql 中。。。")
    log.info("删除mysql里边的数据")
    MysqlUtils.delete(day, lastday);

    val mysqlConf = new MysqlConfiguration

    log.info("数据写入到mysql")
    ds.write.mode("append").format("jdbc").option("url", mysqlConf.mysqlJdbcUrl)
      .option("dbtable", "dws_delivery_in_salor_nc_d")
      .option("user", mysqlConf.dataSourceUser)
      .option("password", mysqlConf.dataSourcePassword)
      .option("driver", mysqlConf.mysqlJdbcDriver)
      .option("batchsize", 10000)
      .save()

  }

  override def write2dws_saleinvoice_in_salor_nc_d(spark: SparkSession, day: String, lastday: String, ds: DataFrame, numPartition: Int, log: Logger): Unit = {

    import spark.implicits._
    log.info("数据写入到mysql 中。。。")
    log.info("删除mysql里边的数据")
    MysqlUtils.delete_dws_saleinvoice_in_salor_nc_d(day, lastday);

    val mysqlConf = new MysqlConfiguration

    log.info("数据写入到mysql")
    ds.write.mode("append").format("jdbc").option("url", mysqlConf.mysqlJdbcUrl)
      .option("dbtable", "dws_saleinvoice_in_salor_nc_d")
      .option("user", mysqlConf.dataSourceUser)
      .option("password", mysqlConf.dataSourcePassword)
      .option("driver", mysqlConf.mysqlJdbcDriver)
      .option("batchsize", 10000)
      .save()

  }


  override def session(): SparkSession = {
    val sparkConf = new SparkConfiguration

    SparkSession.builder
      .appName("OdsApplication")
      .config("hive.exec.dynamic.partition", true)
      .config("hive.exec.dynamic.partition.mode", "nonstrict")
      .config("fs.tos.access.key", sparkConf.ak)
      .config("fs.tos.secret.key", sparkConf.sk)
      .enableHiveSupport().getOrCreate()


  }

  override def write2dws_saleinvoice_in_customer_nc_d(spark: SparkSession, day: String, lastday: String, ds: DataFrame, numPartition: Int, log: Logger): Unit = {

    import spark.implicits._
    log.info("数据写入到mysql 中。。。")
    log.info("删除mysql里边的数据")
    MysqlUtils.delete_dws_saleinvoice_in_customer_nc_d(day, lastday);

    val mysqlConf = new MysqlConfiguration

    log.info("数据写入到mysql")
    ds.write.mode("append").format("jdbc").option("url", mysqlConf.mysqlJdbcUrl)
      .option("dbtable", "dws_saleinvoice_in_customer_nc_d")
      .option("user", mysqlConf.dataSourceUser)
      .option("password", mysqlConf.dataSourcePassword)
      .option("driver", mysqlConf.mysqlJdbcDriver)
      .option("batchsize", 10000)
      .save()

  }

  override def write2dws_delivery_in_industry_nc_d(spark: SparkSession, day: String, lastday: String, ds: DataFrame, numPartition: Int, log: Logger): Unit = {

    import spark.implicits._
    log.info("数据写入到mysql 中。。。")
    log.info("删除mysql里边的数据")
    MysqlUtils.delete_dws_delivery_in_industry_nc_d(day, lastday);

    val mysqlConf = new MysqlConfiguration

    log.info("数据写入到mysql")
    ds.write.mode("append").format("jdbc").option("url", mysqlConf.mysqlJdbcUrl)
      .option("dbtable", "dws_delivery_in_industry_nc_d")
      .option("user", mysqlConf.dataSourceUser)
      .option("password", mysqlConf.dataSourcePassword)
      .option("driver", mysqlConf.mysqlJdbcDriver)
      .option("batchsize", 10000)
      .save()


  }

  override def write2dws_delivery_in_industry_nc_m(spark: SparkSession, day: String, lastday: String, ds: DataFrame, numPartition: Int, log: Logger): Unit = {
    import spark.implicits._
    log.info("数据写入到mysql 中。。。")
    log.info("删除mysql里边的数据")
    MysqlUtils.delete_dws_delivery_in_industry_nc_m(day, lastday);

    val mysqlConf = new MysqlConfiguration

    log.info("数据写入到mysql")
    ds.write.mode("append").format("jdbc").option("url", mysqlConf.mysqlJdbcUrl)
      .option("dbtable", "dws_delivery_in_industry_nc_m")
      .option("user", mysqlConf.dataSourceUser)
      .option("password", mysqlConf.dataSourcePassword)
      .option("driver", mysqlConf.mysqlJdbcDriver)
      .option("batchsize", 10000)
      .save()


  }

  override def write2dwd_saleinvoice_custome_industry_nc_d(spark: SparkSession, day: String, lastday: String, ds: DataFrame, numPartition: Int, log: Logger): Unit = {
    import spark.implicits._
    log.info("数据写入到mysql 中。。。")
    log.info("删除mysql里边的数据")
    MysqlUtils.delete_dwd_saleinvoice_custome_industry_nc_d(day, lastday);

    val mysqlConf = new MysqlConfiguration

    log.info("数据写入到mysql")
    ds.write.mode("append").format("jdbc").option("url", mysqlConf.mysqlJdbcUrl)
      .option("dbtable", "dwd_saleinvoice_custome_industry_nc_d")
      .option("user", mysqlConf.dataSourceUser)
      .option("password", mysqlConf.dataSourcePassword)
      .option("driver", mysqlConf.mysqlJdbcDriver)
      .option("batchsize", 10000)
      .save()
  }

  override def write2dws_tab_fhinfo_hb_nc_m(spark: SparkSession, day: String, lastday: String, ds: DataFrame, numPartition: Int, log: Logger): Unit = {
    import spark.implicits._
    log.info("数据写入到mysql 中。。。")
    log.info("删除mysql里边的数据")
    val month = day.substring(0, 6)
    val lastmonth = lastday.substring(0, 6)
    MysqlUtils.delete_dws_tab_fhinfo_hb_nc_m(month, lastmonth);

    val mysqlConf = new MysqlConfiguration
    val resutDs = ds.repartition(3)

    log.info("数据写入到mysql")
    resutDs.write.mode("append").format("jdbc").option("url", mysqlConf.mysqlJdbcUrl)
      .option("dbtable", "dws_tab_fhinfo_hb_nc_m")
      .option("user", mysqlConf.dataSourceUser)
      .option("password", mysqlConf.dataSourcePassword)
      .option("driver", mysqlConf.mysqlJdbcDriver)
      .option("batchsize", 5000)
      .save()
  }

  override def write2dwd_delivery_nc(spark: SparkSession, day: String, lastday: String, ds: DataFrame, numPartition: Int, log: Logger): Unit = {
    import spark.implicits._
    log.info("数据写入到mysql 中。。。")
    log.info("删除mysql里边的数据")
    MysqlUtils.delete_dwd_delivery_nc(day, lastday);

    val mysqlConf = new MysqlConfiguration
    val resutDs = ds.repartition(3)

    log.info("数据写入到mysql")
    resutDs.write.mode("append").format("jdbc").option("url", mysqlConf.mysqlJdbcUrl)
      .option("dbtable", "dwd_delivery_nc")
      .option("user", mysqlConf.dataSourceUser)
      .option("password", mysqlConf.dataSourcePassword)
      .option("driver", mysqlConf.mysqlJdbcDriver)
      .option("batchsize", 10000)
      .save()

  }

  override def write2dwd_saleinvoice_nc(spark: SparkSession, day: String, lastday: String, ds: DataFrame, numPartition: Int, log: Logger): Unit = {

    import spark.implicits._
    log.info("数据写入到mysql 中。。。")
    log.info("删除mysql里边的数据")
    MysqlUtils.delete_dwd_saleinvoice_nc(day, lastday);

    val mysqlConf = new MysqlConfiguration
    val resutDs = ds.repartition(3)

    log.info("数据写入到mysql")
    resutDs.write.mode("append").format("jdbc").option("url", mysqlConf.mysqlJdbcUrl)
      .option("dbtable", "dwd_saleinvoice_nc")
      .option("user", mysqlConf.dataSourceUser)
      .option("password", mysqlConf.dataSourcePassword)
      .option("driver", mysqlConf.mysqlJdbcDriver)
      .option("batchsize", 10000)
      .save()

  }

  override def write2dws_delivery_in_salor_nc_w(spark: SparkSession, day: String, lastday: String, ds: DataFrame, numPartition: Int, log: Logger): Unit = {
    import spark.implicits._
    log.info("数据写入到mysql 中。。。")
    log.info("删除mysql里边的数据")
    val month = day.substring(0, 6)
    val lastmonth = lastday.substring(0, 6)
    MysqlUtils.delete_dws_delivery_in_salor_nc_w(month, lastmonth);

    val mysqlConf = new MysqlConfiguration
    val resutDs = ds.repartition(3)

    log.info("数据写入到mysql")
    resutDs.write.mode("append").format("jdbc").option("url", mysqlConf.mysqlJdbcUrl)
      .option("dbtable", "dws_delivery_in_salor_nc_w")
      .option("user", mysqlConf.dataSourceUser)
      .option("password", mysqlConf.dataSourcePassword)
      .option("driver", mysqlConf.mysqlJdbcDriver)
      .option("batchsize", 10000)
      .save()


  }


}